顯示具有 python3 標籤的文章。 顯示所有文章
顯示具有 python3 標籤的文章。 顯示所有文章

2025年4月22日 星期二

Python 開發筆記 - 使用 http.server 建置嵌入式產品網頁開發環境與 Proxy 機制

這個需求快十年前弄過,那時是用 webpack proxy server,可以在習慣的桌機開發網頁,接著對於 CGI 查詢就透過 proxy 導向到 device CGI 

然後,現在弄一個 Python ,方便後續在老 server 上運行,畢竟肥肥的程式碼搭配檔案系統大小寫問題,還是交給 server 檔案系統處理吧,如此,未來有部分 html/js/css code 要快速測試,就只需要到指定機器上,運行

$ python3 proxy-server.py --port 12345 --proxy-target 'http://192.168.123.234:80' --proxy-paths '/cgi-bin/' -d project/path/html/code/
Serving files from directory: project/path/html/code/
Serving HTTP on 0.0.0.0 port 12345 (multi-threaded) ...
Proxying paths ['/cgi-bin/'] to http://192.168.123.234:80
^C
Server stopped.

如此對於 CGI request 就會自動導向到 'http://192.168.123.234:80' 

程式碼純筆記,這都是 AI 產的:

#!/usr/bin/env python3

import http.server
import socketserver
import http.client
import argparse
import os
import sys
import socket
from urllib.parse import urlparse

# 使用 ThreadingMixIn 來處理並發請求
class ThreadedHTTPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
    daemon_threads = True
    allow_reuse_address = True


class ProxyHTTPRequestHandler(http.server.SimpleHTTPRequestHandler):
    # 代理目標地址
    proxy_target = None
    # 需要代理的路徑前綴
    proxy_paths = []
    # 連接池(簡單實現)
    conn_pool = {}
    
    def __init__(self, *args, **kwargs):
        # 確保可以正確處理目錄參數
        super().__init__(*args, **kwargs)
    
    def do_GET(self):
        # 檢查是否需要代理這個請求
        for path_prefix in self.proxy_paths:
            if self.path.startswith(path_prefix):
                self.proxy_request()
                return
        
        # 如果不需要代理,則使用默認處理方式
        super().do_GET()
    
    def do_POST(self):
        # 為 POST 請求也提供代理功能
        for path_prefix in self.proxy_paths:
            if self.path.startswith(path_prefix):
                self.proxy_request()
                return
        
        super().do_POST()
    
    def get_connection(self, host, is_ssl=False):
        """從連接池獲取連接,或建立新連接"""
        key = (host, is_ssl)
        
        if key not in self.conn_pool:
            if is_ssl:
                self.conn_pool[key] = http.client.HTTPSConnection(host)
            else:
                self.conn_pool[key] = http.client.HTTPConnection(host)
        
        # 檢查連接是否仍然有效
        conn = self.conn_pool[key]
        try:
            # 嘗試使用一個非阻塞的方式檢查連接狀態
            old_timeout = conn.sock.gettimeout()
            conn.sock.settimeout(0.01)
            conn.sock.recv(1, socket.MSG_PEEK)
            conn.sock.settimeout(old_timeout)
        except (socket.error, AttributeError):
            # 連接已關閉或存在問題,創建新連接
            if is_ssl:
                self.conn_pool[key] = http.client.HTTPSConnection(host)
            else:
                self.conn_pool[key] = http.client.HTTPConnection(host)
        except Exception:
            # 其他類型的錯誤,可能連接仍然有效
            pass
            
        return self.conn_pool[key]
    
    def proxy_request(self):
        """使用 http.client 處理代理請求,這在 Python 3.2 中更高效"""
        target_url = urlparse(self.proxy_target + self.path)
        
        # 獲取請求頭部
        headers = {}
        for key, value in self.headers.items():
            # 排除一些特定的頭部
            if key.lower() not in ('host', 'content-length'):
                headers[key] = value
        
        # 設置正確的 Host 頭部
        headers['Host'] = target_url.netloc
        
        # 讀取請求體(如果有)
        content_length = int(self.headers.get('Content-Length', 0))
        body = self.rfile.read(content_length) if content_length > 0 else None
        
        try:
            # 確定是否使用 HTTPS
            is_ssl = target_url.scheme == 'https'
            
            # 獲取或創建連接
            conn = self.get_connection(target_url.netloc, is_ssl)
            
            # 構建請求路徑
            request_path = target_url.path
            if target_url.query:
                request_path += '?' + target_url.query
            
            # 發送請求
            conn.request(
                method=self.command,
                url=request_path,
                body=body,
                headers=headers
            )
            
            # 獲取響應
            response = conn.getresponse()
            
            # 設置響應狀態碼
            self.send_response(response.status)
            
            # 設置響應頭部
            for header in response.getheaders():
                key, value = header
                if key.lower() != 'transfer-encoding':  # 排除特定頭部
                    self.send_header(key, value)
            self.end_headers()
            
            # 發送響應體
            self.wfile.write(response.read())
            
            # 不關閉連接,將其保留在連接池中
            
        except http.client.HTTPException as e:
            # 處理 HTTP 錯誤
            self.send_response(500)
            self.send_header('Content-Type', 'text/plain; charset=utf-8')
            self.end_headers()
            self.wfile.write("HTTP Error: {}".format(str(e)).encode('utf-8'))
            
        except Exception as e:
            # 處理其他錯誤
            self.send_response(500)
            self.send_header('Content-Type', 'text/plain; charset=utf-8')
            self.end_headers()
            self.wfile.write("Proxy error: {}".format(str(e)).encode('utf-8'))


def run_server(port=8000, proxy_target="http://localhost:8080", proxy_paths=["/cgi-bin/"], directory=None):
    # 設置代理目標和路徑
    ProxyHTTPRequestHandler.proxy_target = proxy_target
    ProxyHTTPRequestHandler.proxy_paths = proxy_paths
    
    # 確保代理目標是有效的 URL
    if not ProxyHTTPRequestHandler.proxy_target.startswith(('http://', 'https://')):
        ProxyHTTPRequestHandler.proxy_target = "http://{}".format(ProxyHTTPRequestHandler.proxy_target)
    
    # 移除代理目標末尾的斜線(如果有)
    if ProxyHTTPRequestHandler.proxy_target.endswith('/'):
        ProxyHTTPRequestHandler.proxy_target = ProxyHTTPRequestHandler.proxy_target[:-1]
    
    # 設置工作目錄
    if directory:
        os.chdir(directory)
        print("Serving files from directory: {}".format(directory))
    else:
        print("Serving files from current directory: {}".format(os.getcwd()))
    
    # 提高 socket 超時時間以處理慢速連接
    socket.setdefaulttimeout(60)
    
    # 建立服務器(使用線程化服務器)
    server_address = ("", port)
    httpd = ThreadedHTTPServer(server_address, ProxyHTTPRequestHandler)
    
    print("Serving HTTP on 0.0.0.0 port {} (multi-threaded) ...".format(port))
    print("Proxying paths {} to {}".format(proxy_paths, proxy_target))
    
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\nServer stopped.")
        httpd.server_close()


if __name__ == "__main__":
    # 解析命令行參數
    parser = argparse.ArgumentParser(description='HTTP Server with proxy capabilities (optimized for Python 3.2)')
    parser.add_argument('--port', type=int, default=8000, help='Port to listen on (default: 8000)')
    parser.add_argument('--proxy-target', type=str, default="http://localhost:8080", 
                        help='Target server to proxy to (default: http://localhost:8080)')
    parser.add_argument('--proxy-paths', type=str, default="/cgi-bin/", 
                        help='Comma-separated list of path prefixes to proxy (default: /cgi-bin/)')
    parser.add_argument('--directory', '-d', type=str, default=None,
                        help='Specify directory to serve files from (default: current directory)')
    
    args = parser.parse_args()
    proxy_paths = [path.strip() for path in args.proxy_paths.split(',')]
    
    run_server(args.port, args.proxy_target, proxy_paths, args.directory)

2024年8月23日 星期五

Python 開發筆記 - 不透過 Google API Key 下載公開 Google Sheets 資料,將每個 Sheet 匯出 csv 格式

有個工作任務要做 Google Sheets 資料比對,最簡單的方式就把他們匯出後用 git diff 來比對即可,想試著用 AI 產生一隻 python 小工具,只要輸入 Google Sheets URL 或是 Google Sheets URL 內關鍵的辨識 ID (在此稱作 spreadsheet id),就能夠下載該 Google Spreadsheet 內所有 sheet 資料

然後,要下載指定 sheet 必須得知每個 sheet gid ,這個問 ChatGPT-4o 或 Claude.ai 老半天還是沒法解,包過上傳 html static code,最後自己還是跳下來收尾人工刻一下,原理:

  • 先設法下載到 HTML Code
  • 透過 docs-sheet-tab-caption 抓出 Sheet Name
  • 透過 var bootstrapData = {...}; 得知內有 Sheet Name 與 Gid 的資料
  • 再用 [0,0,\"gid\",[ 格式,找到 gid
連續動作:

% python3 main.py 

usage: main.py [-h] (--google-spreadsheet-url GOOGLE_SPREADSHEET_URL | --google-spreadsheet-id GOOGLE_SPREADSHEET_ID) [--output OUTPUT]

main.py: error: one of the arguments --google-spreadsheet-url --google-spreadsheet-id is required


% python3 main.py --google-spreadsheet-id 'XXXXXXXXXXXXXXXXXXXXXXX'

[INFO] Downloaded sheet: sheet01 to sheets_csv/sheet01.csv

[INFO] Downloaded sheet: sheet02 to sheets_csv/sheet02.csv

[INFO] Downloaded sheet: sheet03 to sheets_csv/sheet03.csv

[INFO] Downloaded sheet: sheet04 to sheets_csv/sheet04.csv

[INFO] Downloaded sheet: sheet05 to sheets_csv/sheet05.csv


程式碼:

```
% cat main.py 
import argparse
import re
import requests
import json
import os

def extract_spreadsheet_id(url):
    match = re.search(r'/d/([a-zA-Z0-9-_]+)', url)
    return match.group(1) if match else None

def get_spreadsheet_info(spreadsheet_id):
    url = f"https://docs.google.com/spreadsheets/d/{spreadsheet_id}/edit"
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
    }
    
    try:
        response = requests.get(url, headers=headers)
        response.raise_for_status()
        html_content = response.text
        
        # Extract sheet names
        sheet_names = re.findall(r'docs-sheet-tab-caption[^>]+>([^<]+)</div>', html_content)
        #print(f"Found sheet names: {sheet_names}")

        # Extract mergedConfig
        config_match = re.search(r'var bootstrapData\s*=\s*({.*?});', html_content, re.DOTALL)
        if config_match:
            config_str = config_match.group(1)
            sheet_info = {}
            try:
                for index, sheet_name in enumerate(sheet_names):
                    #print(f"Processing sheet: {sheet_name}, index: {index}")
                    beginPattern = f'[{index},0,\\"'
                    endPattern = f'\\",['
                    beginIndex = config_str.find(beginPattern)
                    endIndex = config_str.find(endPattern, beginIndex)
                    gidValue = config_str[beginIndex + len(beginPattern):endIndex]
                    sheet_info[sheet_name] = gidValue
                return sheet_info
            except Exception as e:
                print(f"[INFO] Error extracting sheet information: {e}")
                return None
        else:
            print("[INFO] Could not find bootstrapData in the HTML content")
            return None
    except requests.RequestException as e:
        print(f"[INFO] Error fetching the spreadsheet: {e}")
        return None

def download_sheet_as_csv(spreadsheet_id, sheet_name, gid, output_folder):
    csv_url = f"https://docs.google.com/spreadsheets/d/{spreadsheet_id}/export?format=csv&gid={gid}"
    csv_response = requests.get(csv_url)
    
    if csv_response.status_code == 200:
        output_path = os.path.join(output_folder, f"{sheet_name}.csv")
        with open(output_path, 'wb') as f:
            f.write(csv_response.content)
        print(f"[INFO] Downloaded sheet: {sheet_name} to {output_path}")
    else:
        print(f"[INFO] Failed to download sheet: {sheet_name}. Status code: {csv_response.status_code}")

def main():
    parser = argparse.ArgumentParser(description="Extract Google Spreadsheet information")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--google-spreadsheet-url", help="Google Spreadsheet URL")
    group.add_argument("--google-spreadsheet-id", help="Google Spreadsheet ID")
    parser.add_argument('--output', type=str, default='sheets_csv', help='The directory to save the CSV files')
    args = parser.parse_args()

    if args.google_spreadsheet_url:
        spreadsheet_id = extract_spreadsheet_id(args.google_spreadsheet_url)
    else:
        spreadsheet_id = args.google_spreadsheet_id

    if not spreadsheet_id:
        print("[INFO] Invalid Google Spreadsheet URL or ID")
        return

    sheet_info = get_spreadsheet_info(spreadsheet_id)
    if sheet_info:
        for name, gid in sheet_info.items():
            download_sheet_as_csv(spreadsheet_id, name, gid, args.output)
    else:
        print("[INFO] Failed to extract sheet information")

if __name__ == "__main__":
    main()
```

2024年8月2日 星期五

Node.js 開發筆記 - 分別透過 Pyodide, Brython, WebAssembly 在 node.js 呼叫 Python Code @ node.js v20, python3.11

一時興起研究一下 node.js 呼叫 python code 的方式,當然,都在 linux server 可以直接用 child_process 直接呼叫 python 去運行,例如 nodejs.org/api/child_process.html 的範例

```
const { spawn } = require('node:child_process');
const ls = spawn('ls', ['-lh', '/usr']);

ls.stdout.on('data', (data) => {
  console.log(`stdout: ${data}`);
});

ls.stderr.on('data', (data) => {
  console.error(`stderr: ${data}`);
});

ls.on('close', (code) => {
  console.log(`child process exited with code ${code}`);
}); 
```

然而,有沒有可能在 node.js 內,直接做 python 直譯的過程等方式?當個樂子,找了一下,還真的有,這樣搞的優勢是降低環境部署的變因,當然,效率上不見得是好辦法,但可以讓不同語言的開發者進行融合(誤),目前看到兩種整合方式:
其中 Brython 屬於設計在 Web Browser 下運行(需要 DOM 資源),而 Pyodide 則不需要。分別筆記一下用法。

首先是要執行的 python code 內有 python 的 re 跟 json 模組的使用:

```
% cat script.py 
import re
import json

def runTest(inputData):
    output = {}
    flags = 0
    pattern = r'''(?x)
      (?:
        \.get\("n"\)\)&&\(b=|
        (?:
          b=String\.fromCharCode\(110\)|
          (?P<str_idx>[a-zA-Z0-9_$.]+)&&\(b="nn"\[\+(?P=str_idx)\]
        ),c=a\.get\(b\)\)&&\(c=|
        \b(?P<var>[a-zA-Z0-9_$]+)=
      )(?P<nfunc>[a-zA-Z0-9_$]+)(?:\[(?P<idx>\d+)\])?\([a-zA-Z]\)
      (?(var),[a-zA-Z0-9_$]+\.set\("n"\,(?P=var)\),(?P=nfunc)\.length)'''

    try:
        result = re.search(pattern, inputData, flags)
        if result:
            output["status"] = True
            output["data"] = result.groupdict()
    except Exception as e:
        output["error"] = str(e)
    return json.dumps(output, indent=4)

runTest(data)
```

Pyodide 用法:

```
% nvm use v20
Now using node v20.10.0 (npm v10.2.3)
% npm install pyodide
% cat package.json 
{
  "dependencies": {
    "pyodide": "^0.26.2"
  }
}

% cat run.js
const fs = require('fs').promises;
const { loadPyodide } = require("pyodide");

async function main() {
  const fileContent = await fs.readFile('mydata.bin', 'utf8');
  let pyodide = await loadPyodide();
  pyodide.globals.set("data", pyodide.toPy(fileContent));
  const pythonCode = await fs.readFile('script.py', 'utf8');
  let result = pyodide.runPython(pythonCode);
  console.log(result);
}

main();

% echo "Hello World" > mydata.bin

% node run.js
{}
```

上述使用過程算直觀,但偷懶把要傳遞的資料設定在全域變數,在用 node.js 環境接住 python 運算的結果,看來這個效果是很 OK ,有正常運行得到期待的結果。

接著研究 Brython 用法,他設計上需要 Browser 環境做事:

% head -n 9 brython.js
// brython.js brython.info
// version [3, 11, 0, 'final', 0]
// implementation [3, 11, 3, 'dev', 0]
// version compiled from commented, indented source files at
// github.com/brython-dev/brython
var __BRYTHON__=__BRYTHON__ ||{}
try{
eval("async function* f(){}")}catch(err){console.warn("Your browser is not fully supported. If you are using "+
"Microsoft Edge, please upgrade to the latest version")}

在 node.js 需要 jsdom 模擬一些環境,而下載 brython.js 和 brython_stdlib.js 則參考官網文件,透過 pip install brython 工具出來使用,所以這邊的流程會多了 python 工具的安裝,且現況用 python 3.12 會顯示有些問題,就先定在 3.11 版。此外 brython.js 運行環境,也是可以用最新版 node.js v22 ,但是會看到 [DEP0040] DeprecationWarning: The `punycode` module is deprecated. Please use a userland alternative instead. 訊息,所以先退到 node.js v20 避免額外的訊息

連續動作:

```
% python3.11 -m venv venv
% source venv/bin/activate
(venv) % pip install brython
Collecting brython
  Using cached brython-3.11.3-py3-none-any.whl.metadata (1.0 kB)
Using cached brython-3.11.3-py3-none-any.whl (1.6 MB)
Installing collected packages: brython
Successfully installed brython
(venv) % brython-cli install
Installing Brython 3.11.3
done
(venv) % ls
README.txt brython_stdlib.js index.html venv
brython.js demo.html unicode.txt
```

接著回到 node.js 主場:

```
% nvm use v20
Now using node v20.10.0 (npm v10.2.3)
% cat package.json 
{
  "dependencies": {
    "jsdom": "^24.1.1"
  }
}
% cat run-via-dom.js 
const { JSDOM } = require('jsdom');
const fs = require('fs');
const path = require('path');

const dom = new JSDOM(`<!DOCTYPE html>
<html>
    <head></head>
    <body>
        <script></script>
    </body>
</html>`, {
    runScripts: "dangerously", 
    resources: "usable"
});

const brythonJsPath = path.join(__dirname, 'brython.js');
const brythonStdlibJsPath = path.join(__dirname, 'brython_stdlib.js');

const brythonJs = fs.readFileSync(brythonJsPath, 'utf8');
const brythonStdlibJs = fs.readFileSync(brythonStdlibJsPath, 'utf8');

try {
    dom.window.eval(brythonJs);
    dom.window.eval(brythonStdlibJs);
} catch (error) {
    console.error('Error executing brython.js:', error);
}

const scriptPath = path.join(__dirname, 'script.py');
const pythonScript = fs.readFileSync(scriptPath, 'utf8');

const dataPath = path.join(__dirname, 'mydata.bin');
const binaryData = fs.readFileSync(dataPath, 'utf8');
const base64Data = Buffer.from(binaryData).toString('base64');

const scriptElement = dom.window.document.createElement('script');
scriptElement.type = 'text/python';
scriptElement.textContent = `
import base64
data = base64.b64decode("""${base64Data}""")

${pythonScript}

from browser import document
document.output = runTest(data)
`
dom.window.document.body.appendChild(scriptElement);
try {
    dom.window.brython({debug: 1, pythonpath: ['.']})
    console.log(dom.window.document.output);
} catch (error) {
    console.error('Error executing dom.window.brython:', error);
}
console.log('Python script execution completed.');

% echo "Hello World" > mydata.bin

% node run-via-dom.js 
{"status": false, "data": {}, "error": "not the same type for string and pattern"}
Python script execution completed.
```

很可惜的,剛好要實驗複雜的 python regular expression,在 brython.js + node.js v20 + jsdom 環境上失敗了,甚至小改 index.html 搭配 python3 -m http.server 用 Chrome browser 執行(給予他完整的 Chrome 瀏覽器環境)還是有一樣的錯誤訊息,這邊就暫時推論失敗了,而上述的範例已經包括從 node.js 傳資料到 python code ,以及運行完如何把回傳資料傳到 node.js 使用,眼尖的人,應該會發現在 brython 用法內,使用了 `document.output = runTest(data)` ,其實是多呼叫了一次 runTest(data),因為原先 `${pythonScript}` 也有做,但沒在細追怎樣接運算結果,剛好不合預期就放棄研究。

最後,就是 WebAssembly 領域(一開始寫這篇筆記就是要研究 WebAssembly ,不小心走偏),把某一種 python code 轉成 wasm 格式,接著用 wasmer 運行,或是在其他語言(如 node.js)運行 wasm code。

先透過 MacPorts 安裝 wasmer:

% port search wasmer
wasmer @4.3.5 (lang, devel)
    The leading WebAssembly Runtime supporting WASI and Emscripten
% sudo port install wasmer

接著試著用 py2wasm 把 script-main.py 轉成 script-main.wasm,其中 py2wasm 官網有提到目前僅支援 python3.11:

% python3.11 -m venv venv
% source venv/bin/activate
(venv) % pip install py2wasm
(venv) % py2wasm script-main.py -o script-main.wasm

程式碼:

```
% cat script-main.py
import re
import json

def runTest(inputData):
    output = { "status": False, "data": {}, "error": None}
    flags = 0
    pattern = r'''(?x)
      (?:
        \.get\("n"\)\)&&\(b=|
        (?:
          b=String\.fromCharCode\(110\)|
          (?P<str_idx>[a-zA-Z0-9_$.]+)&&\(b="nn"\[\+(?P=str_idx)\]
        ),c=a\.get\(b\)\)&&\(c=|
        \b(?P<var>[a-zA-Z0-9_$]+)=
      )(?P<nfunc>[a-zA-Z0-9_$]+)(?:\[(?P<idx>\d+)\])?\([a-zA-Z]\)
      (?(var),[a-zA-Z0-9_$]+\.set\("n"\,(?P=var)\),(?P=nfunc)\.length)'''

    try:
        result = re.search(pattern, inputData, flags)
        if result:
            output["status"] = True
            output["data"] = result.groupdict()
    except Exception as e:
        output["error"] = str(e)
    return json.dumps(output, indent=4)

if __name__ == "__main__":
    import sys
    if len(sys.argv) < 2:
        print("Usage: python script-main.py <inputData>")
    else:
        print(runTest(sys.argv[1]))

% python3 script-main.py 
Usage: python script-main.py <inputData>

% python3 script-main.py "Hello World"
{
    "status": false,
    "data": {},
    "error": null
}
```

wasmer 實測:

```
% wasmer run script-main.wasm 
Usage: python script-main.py <inputData>

% wasmer run script-main.wasm "Hello World"
{
    "status": false,
    "data": {},
    "error": null
}
```

接著讓 Node.JS 來運行,這邊就來煩 ChatGPT 並小改一下,有了一個比較堪用的版本:

```
% cat run.js 
const fs = require('fs');
const { WASI } = require('wasi');
const path = require('path');
const { TextDecoder } = require('util');

const runWasm = async (inputData) => {
    const wasmPath = path.resolve('./script-main.wasm');
    const wasmBinary = fs.readFileSync(wasmPath);

    // Setup a WASI instance
    const wasi = new WASI({
        args: inputData ? ['script-main.wasm', inputData] : ['script-main.wasm'],
        env: {},
        version: 'preview1'
    });

    // Create a memory buffer for the stdout
    const memory = new WebAssembly.Memory({ initial: 1 });

    // Compile and instantiate the WebAssembly module
    const { instance } = await WebAssembly.instantiate(wasmBinary, {
        wasi_snapshot_preview1: wasi.wasiImport,
        env: { memory }
    });

    // Start the WASI instance
    wasi.start(instance);

    // Read and decode the stdout data
    const stdout = new Uint8Array(memory.buffer);
    const decoder = new TextDecoder('utf8');
    const output = decoder.decode(stdout);
    console.log(output.trim());
};

// Get the inputData from the command line arguments
runWasm(process.argv[2] || null).catch(console.error);

% nvm use v20
Now using node v20.10.0 (npm v10.2.3)

% node run.js 
(node:22046) ExperimentalWarning: WASI is an experimental feature and might change at any time
(Use `node --trace-warnings ...` to show where the warning was created)
Usage: python script-main.py <inputData>

% node run.js "Hello World"
(node:22050) ExperimentalWarning: WASI is an experimental feature and might change at any time
(Use `node --trace-warnings ...` to show where the warning was created)
{
    "status": false,
    "data": {},
    "error": null
}
```

回過頭來,故事起源是想善用一些 open source 甚至不同程式語言的整合架構,因此稍微研究一些跨語言的整合,很可惜的,最佳的路線應當還是各自跑在各自的 runtime 環境,以上就當趣味筆記一下。

2023年12月25日 星期一

Python 開發筆記 - 開發 epub-image-helper 工具,批次將圖片目錄打包成 EPUB 格式,也支援 PDF to EPUB @ macos 14.2.1, Python 3.11

記得剛出社會沒多久,被派去研究電子書領域,那時就接觸了 EPUB 格式。當時在團隊中被分配到開發 EPUB READER,例如在 HTML5 環境中,在 Browser 讀進 epub 檔案、進行 unzip 、抽出目錄、呈現書的內容。後來也做過電子書相關的應用發想,而主管們相繼離開(?),也讓我接手和接觸了貢獻 webkit 的項目(C++),可惜很多狀態都不了了之。

大概兩三週前想到 EPUB 這格式,想嘗試把照片打包成單一 EPUB 檔案,方便收藏。但大概完成七八成後,才發現有很多限制,例如電子書閱讀器上要處理大檔案時,就會陷入到電子書閱讀器的硬體規格和處理能力,甚至它實作處理 EPUB 格式的方式等,我才發現當初想要把一堆照片合成一大本應當不是正道 XD 後來還是順手練習 python open source 工具,做了一些發布,在此就筆記一下。而能妥善處理大檔案的電子書閱讀器,大概就 iOS devices 而已,實際在 iPhone SE3 上處理 500MB 的電子書EPUB也都很 OK 的。

這次挑選 ebooklib 作為 EPUB 製作的核心函式庫,當初查看 github 專案上有破千的關注,跳下去使用時才發現有些限制,也順手發了兩次 pull requests ,查閱了該專案歷年的 pull requests 活躍狀態,還是自己弄了個 ebooklib.changyy 方便自己整合

最後產出的小專案:
安裝 epub-image-helper:

% pip install epub-image-helper
% epub-image-helper | jq -r '.version'
1.0.4

使用方式可以參考 README 或是 https://github.com/changyy/epub-image-helper/tree/main/example 內的小程式庫,主要可以用 epub-image-helper 將有階層式的照片打包起來,也可以自訂一個 input.json 格式,批次把許多目錄依照一些規則打包起來,如:

% cd example/01-batch-image-to-epub
% tree .
.
├── epub
├── image2epub.py
├── input.json
└── storage
    └── url 
        ├── 001 
        │   ├── 01.jpg
        │   ├── 02.jpg
        │   └── 03.jpg
        └── 002 
            ├── 01.jpg
            ├── 02.jpg
            ├── 03.jpg
            ├── 04.jpg
            └── 05.jpg

6 directories, 10 files
% cat input.json
[
  {   
    "name": "MyPhotosA",
    "author": ["Author"],
    "books": [
      "https://localhost/001/A", 
      "https://localhost/002/B", 
      "https://localhost/003/C"
    ]   
  },  
  {   
    "name": "MyPhotosB",
    "author": ["Author"],
    "books": [
      "https://localhost/004/A", 
      "https://localhost/005/B", 
      "https://localhost/006/C"
    ]   
  }
]
% python3 image2epub.py
...
% tree epub 
epub
└── MyPhotosA
    ├── MyPhotosA01.epub
    └── MyPhotosA02.epub

2 directories, 2 files

除此之外,也支援拿 pdf 作為處理,例如把一份 pdf 內所有的圖片都抽出來打包成 EPUB 格式。

估計就告一個段落吧!這個 epub-image-helper 工具發展過程中,發現自己原先想要做的(弄成一包很大的EPUB)並非主流,就漸漸失去熱情。趁熱情還沒散去之前,先弄成一個小 open source 紀念這段2023年12月初,斷斷續續兩個週末的小產出。

2021年10月26日 星期二

Python 筆記 - 使用 AdSense / AdMob API 製作 Dashboard

約莫經營流量變現三年多,大部分的工作都是靠 adsense/admob 網站就夠用了,但如果要整合到其他儀表板的話,就必須靠 AdMob/AdSense API 來撈點資料。

原先一直只靠 AdSense API 做事,但是舊版 AdSense API 在 2021-10-12 下線了,接著新版 AdSense API 就不在有 AdMob 數據,因此需要分開存取。

筆記瑣事:
  • 使用 Python3 開發,並安裝套件相關套件
    • google-api-python-client
    • google-auth-oauthlib
  • AdSense 跟 AdMob 授權的 Scope 是分開的
    • 'https://www.googleapis.com/auth/admob.readonly'
    • 'https://www.googleapis.com/auth/adsense.readonly'
  • Google 專案必須分別啟用 AdSense Management api 跟 AdMob api 使用權
  • Google 專案的憑證,只需使用 "OAuth 2.0 用戶端 ID" ,並採用 "電腦版應用程式" 即可
  • AdSense 跟 AdMob API 輸出的 DATE 欄位的格式不同
    • "2021-10-26"
    • "20211026"
因此,若要符合之前舊版 Adsense 的輸出,需要分別使用 AdSense api v2 跟 AdMob API v1 來整理數據,例如先將 AdMob 收益用日期作為關鍵字紀錄,接著輸出 AdSense 來源時,在計算當日營收時,再順便查表累計 AdMob 收益。

完整程式碼:

    result = service.accounts().reports().generate(
        account=account_id,
        dateRange='CUSTOM',
        startDate_year=startDate['year'],
        startDate_month=startDate['month'],
        startDate_day=startDate['day'],
        endDate_year=endDate['year'],
        endDate_month=endDate['month'],
        endDate_day=endDate['day'],
        metrics=output['data']['metrics'],
        dimensions=output['data']['dimensions'],
        currencyCode="USD",
        reportingTimeZone="ACCOUNT_TIME_ZONE",
    ).execute()

AdMob API 片段程式碼:

    result = service.accounts().networkReport().generate(
        parent=account_id,
        body= {
            'reportSpec': {
                'dateRange' : {
                    'startDate': output['data']['startDate'],
                    'endDate': output['data']['endDate'],
                },
                'dimensions': output['data']['dimensions'],
                'metrics': output['data']['metrics'],
                'localizationSettings': {
                    'currencyCode': 'USD',
                    'languageCode': 'en-US',
                },
            },
        },
    ).execute()