435 lines
16 KiB
Python
Executable File
435 lines
16 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
# ==============================================================================
|
||
# 用途：Gitea Wiki 讀寫工具（worklog 專用）。提供 token 解析、頁面讀取、
|
||
# 建立、append 追加（read-modify-write + 寫後驗證重試），供 worklog.sh
|
||
# 與 /jsc:worklog skill 共用，避免兩份實作漂移。
|
||
# 更新時間：2026/07/27 11:14:16
|
||
# 相依：Python 3 標準庫（urllib、base64、json、re）。不需 requests、不需 jq。
|
||
# 機密：token 一律從環境變數或本機憑證檔讀取，絕不輸出、絕不寫入任何檔案。
|
||
# ==============================================================================
|
||
|
||
import base64
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import ssl
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
TAIPEI = timezone(timedelta(hours=8))
|
||
|
||
|
||
def _ssl_context():
|
||
"""
|
||
建立 TLS 連線設定：維持完整憑證鏈驗證，僅關閉 VERIFY_X509_STRICT。
|
||
|
||
Python 3.13 起預設啟用 X509 嚴格檢查，內部 CA 憑證若缺少 Subject Key
|
||
Identifier 會被拒絕（curl 不做此檢查，故 curl 可連而 Python 不行）。
|
||
此處只放寬擴充欄位的嚴格檢查，主機名稱與憑證鏈驗證仍完整保留。
|
||
"""
|
||
ctx = ssl.create_default_context()
|
||
ctx.verify_flags &= ~ssl.VERIFY_X509_STRICT
|
||
return ctx
|
||
|
||
|
||
SSL_CONTEXT = _ssl_context()
|
||
|
||
|
||
def now_str():
|
||
"""取得台灣時區的 yyyy/MM/dd HH:mm:ss 時間字串。"""
|
||
return datetime.now(TAIPEI).strftime("%Y/%m/%d %H:%M:%S")
|
||
|
||
|
||
def log(level, message, stage="wiki_api"):
|
||
"""輸出統一格式訊息（[時間][階段][等級]: 訊息，一行一則），一律走 stderr 不污染 stdout。"""
|
||
print(f"[{now_str()}][{stage}][{level}]: {message}", file=sys.stderr)
|
||
|
||
|
||
def mask(text, secret):
|
||
"""將字串中的 secret 遮蔽為 ***，避免 token 洩漏到輸出。"""
|
||
if not secret:
|
||
return text
|
||
return text.replace(secret, "***")
|
||
|
||
|
||
# ------------------------------------------------------------------------------
|
||
# token 解析：GITEA_TOKEN → tea config → git-credentials
|
||
# ------------------------------------------------------------------------------
|
||
|
||
def _token_from_tea(host):
|
||
"""從 tea 設定檔取出指定 host 的 token（找不到回 None）。"""
|
||
for path in ("~/.config/tea/config.yml", "~/.tea/config.yml"):
|
||
f = os.path.expanduser(path)
|
||
if not os.path.isfile(f):
|
||
continue
|
||
try:
|
||
raw = open(f, encoding="utf-8").read()
|
||
except OSError:
|
||
continue
|
||
for block in re.split(r"(?m)^\s*-\s+name:", raw):
|
||
if host not in block:
|
||
continue
|
||
m = re.search(r"(?m)^\s*token:\s*[\"']?([A-Za-z0-9_\-]+)", block)
|
||
if m:
|
||
return m.group(1)
|
||
return None
|
||
|
||
|
||
def _token_from_git_credentials(host):
|
||
"""從 ~/.git-credentials（credential.helper=store）取出指定 host 的密碼作為 token。"""
|
||
f = os.path.expanduser("~/.git-credentials")
|
||
if not os.path.isfile(f):
|
||
return None
|
||
try:
|
||
lines = open(f, encoding="utf-8").read().splitlines()
|
||
except OSError:
|
||
return None
|
||
for line in lines:
|
||
m = re.match(r"https?://([^:]+):([^@]+)@(.+)$", line.strip())
|
||
if m and m.group(3) == host:
|
||
return urllib.parse.unquote(m.group(2))
|
||
return None
|
||
|
||
|
||
def resolve_token(host, repo):
|
||
"""
|
||
依固定優先序解析可用 token，並以 GET /repos/<repo> 實際驗證權限。
|
||
|
||
優先序：GITEA_TOKEN → tea 設定檔該 host 的 token → ~/.git-credentials。
|
||
回傳 (token, 來源說明)；全部失敗回 (None, 說明)。
|
||
"""
|
||
candidates = []
|
||
env = os.environ.get("GITEA_TOKEN")
|
||
if env:
|
||
candidates.append((env, "GITEA_TOKEN"))
|
||
tea = _token_from_tea(host)
|
||
if tea and tea != env:
|
||
candidates.append((tea, "tea 設定檔"))
|
||
cred = _token_from_git_credentials(host)
|
||
if cred and cred not in (env, tea):
|
||
candidates.append((cred, "git-credentials"))
|
||
|
||
if not candidates:
|
||
return None, "找不到任何可用憑證來源"
|
||
|
||
for token, source in candidates:
|
||
code, _ = _request("GET", f"https://{host}/api/v1/repos/{repo}", token, None)
|
||
if code == 200:
|
||
return token, source
|
||
log("DBG", f"{source} 對 {host} 驗證失敗（HTTP {code}），改試下一個來源")
|
||
return None, f"{len(candidates)} 個憑證來源全部驗證失敗"
|
||
|
||
|
||
# ------------------------------------------------------------------------------
|
||
# HTTP
|
||
# ------------------------------------------------------------------------------
|
||
|
||
def _request(method, url, token, payload):
|
||
"""發出 Gitea API 請求，回傳 (HTTP 狀態碼, 回應內文字串)。網路層錯誤以 0 表示。"""
|
||
data = json.dumps(payload, ensure_ascii=False).encode("utf-8") if payload is not None else None
|
||
req = urllib.request.Request(url, data=data, method=method)
|
||
req.add_header("Authorization", f"token {token}")
|
||
req.add_header("Accept", "application/json")
|
||
if data:
|
||
req.add_header("Content-Type", "application/json")
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=30, context=SSL_CONTEXT) as resp:
|
||
return resp.status, resp.read().decode("utf-8", "replace")
|
||
except urllib.error.HTTPError as e:
|
||
return e.code, e.read().decode("utf-8", "replace")
|
||
except Exception as e: # 網路錯誤、逾時
|
||
return 0, str(e)
|
||
|
||
|
||
def _api_base(host, repo):
|
||
"""組出 repo 層級的 wiki API base URL。"""
|
||
return f"https://{host}/api/v1/repos/{repo}/wiki"
|
||
|
||
|
||
# ------------------------------------------------------------------------------
|
||
# wiki 操作
|
||
# ------------------------------------------------------------------------------
|
||
|
||
def list_pages(host, repo, token):
|
||
"""
|
||
列出 wiki 全部頁面（分頁完整讀取），回傳 (狀態, 頁面清單)。
|
||
|
||
狀態為 'ok'／'missing'（wiki 尚未初始化）／'error'。清單元素含 title 與 sub_url。
|
||
"""
|
||
pages = []
|
||
page_no = 1
|
||
limit = 50
|
||
while True:
|
||
url = f"{_api_base(host, repo)}/pages?page={page_no}&limit={limit}"
|
||
code, body = _request("GET", url, token, None)
|
||
if code == 404:
|
||
return "missing", []
|
||
if code != 200:
|
||
return "error", []
|
||
try:
|
||
batch = json.loads(body)
|
||
except ValueError:
|
||
return "error", []
|
||
if not isinstance(batch, list):
|
||
return "error", []
|
||
pages.extend(batch)
|
||
if len(batch) < limit:
|
||
return "ok", pages
|
||
page_no += 1
|
||
|
||
|
||
def resolve_sub_url(host, repo, token, title):
|
||
"""
|
||
以 title 查出 Gitea 實際的 sub_url。
|
||
|
||
Gitea wiki 會對 title 做轉義（`-` 代表空格，實際 dash 另有轉義形式，例如
|
||
title `Worklog-2026-07-W4` 的 sub_url 為 `Worklog-2026-07-W4.-`），因此讀寫
|
||
一律以查表得到的 sub_url 為準，不自行猜測轉義規則。
|
||
找不到回 None。
|
||
"""
|
||
status, pages = list_pages(host, repo, token)
|
||
if status != "ok":
|
||
return None
|
||
for item in pages:
|
||
if item.get("title") == title:
|
||
return item.get("sub_url") or title
|
||
return None
|
||
|
||
|
||
def get_page(host, repo, token, page):
|
||
"""
|
||
讀取 wiki 頁面內容（page 可傳 title 或 sub_url，內部會自動解析）。
|
||
|
||
回傳 (狀態, 內容字串)；狀態為 'ok'（存在）、'missing'（404，頁面或 wiki 尚未建立）、
|
||
'error'（其他失敗，內容為遮蔽後的錯誤訊息）。
|
||
"""
|
||
sub_url = resolve_sub_url(host, repo, token, page) or page
|
||
url = f"{_api_base(host, repo)}/page/{urllib.parse.quote(sub_url)}"
|
||
code, body = _request("GET", url, token, None)
|
||
if code == 404:
|
||
return "missing", ""
|
||
if code != 200:
|
||
return "error", mask(f"HTTP {code} {body[:200]}", token)
|
||
try:
|
||
data = json.loads(body)
|
||
except ValueError:
|
||
return "error", "回應不是合法 JSON"
|
||
raw = data.get("content_base64") or ""
|
||
try:
|
||
return "ok", base64.b64decode(raw).decode("utf-8", "replace")
|
||
except Exception:
|
||
return "error", "content_base64 解碼失敗"
|
||
|
||
|
||
def create_page(host, repo, token, page, content, message):
|
||
"""建立新的 wiki 頁面（wiki 尚未初始化時亦由此初始化）。回傳 (是否成功, 訊息)。"""
|
||
url = f"{_api_base(host, repo)}/new"
|
||
payload = {
|
||
"title": page,
|
||
"content_base64": base64.b64encode(content.encode("utf-8")).decode("ascii"),
|
||
"message": message,
|
||
}
|
||
code, body = _request("POST", url, token, payload)
|
||
if code in (201, 200):
|
||
return True, f"已建立頁面 {page}"
|
||
return False, mask(f"建立頁面失敗 HTTP {code} {body[:200]}", token)
|
||
|
||
|
||
def delete_page(host, repo, token, page):
|
||
"""刪除 wiki 頁面（page 可傳 title 或 sub_url）。回傳 (是否成功, 訊息)。"""
|
||
sub_url = resolve_sub_url(host, repo, token, page) or page
|
||
url = f"{_api_base(host, repo)}/page/{urllib.parse.quote(sub_url)}"
|
||
code, body = _request("DELETE", url, token, None)
|
||
if code in (204, 200):
|
||
return True, f"已刪除頁面 {page}"
|
||
return False, mask(f"刪除頁面失敗 HTTP {code} {body[:200]}", token)
|
||
|
||
|
||
def update_page(host, repo, token, page, content, message):
|
||
"""整頁覆寫既有 wiki 頁面（append 由呼叫端先合併內容）。回傳 (是否成功, 訊息)。"""
|
||
sub_url = resolve_sub_url(host, repo, token, page) or page
|
||
url = f"{_api_base(host, repo)}/page/{urllib.parse.quote(sub_url)}"
|
||
payload = {
|
||
"title": page,
|
||
"content_base64": base64.b64encode(content.encode("utf-8")).decode("ascii"),
|
||
"message": message,
|
||
}
|
||
code, body = _request("PATCH", url, token, payload)
|
||
if code in (200, 201):
|
||
return True, f"已更新頁面 {page}"
|
||
return False, mask(f"更新頁面失敗 HTTP {code} {body[:200]}", token)
|
||
|
||
|
||
def append_entry(host, repo, token, page, header, entry, marker, retries=3):
|
||
"""
|
||
將條目追加到週頁尾端：讀取現有內容 → 合併 → 寫回 → 寫後讀取驗證。
|
||
|
||
marker 為條目內唯一字串（時間戳＋session 短碼），用於驗證自己的內容確實落地；
|
||
多個 session 同時寫入時，驗證失敗會重讀最新內容重試，避免互相覆蓋。
|
||
回傳 (是否成功, 訊息)。
|
||
"""
|
||
for attempt in range(1, retries + 1):
|
||
status, current = get_page(host, repo, token, page)
|
||
if status == "error":
|
||
return False, f"讀取頁面失敗：{current}"
|
||
|
||
if status == "missing":
|
||
content = f"{header}\n\n{entry}\n"
|
||
ok, msg = create_page(host, repo, token, page, content, f"worklog: 建立 {page}")
|
||
if not ok:
|
||
# wiki 已存在但頁面不存在時，建立可能失敗；下一輪改走更新
|
||
log("WRN", f"第 {attempt} 次建立失敗：{msg}")
|
||
continue
|
||
else:
|
||
if marker in current:
|
||
return True, "條目已存在，無需重複寫入"
|
||
body = current.rstrip("\n")
|
||
if not body:
|
||
body = header
|
||
content = f"{body}\n\n{entry}\n"
|
||
ok, msg = update_page(host, repo, token, page, content, f"worklog: 追加 {marker}")
|
||
if not ok:
|
||
log("WRN", f"第 {attempt} 次寫入失敗：{msg}")
|
||
continue
|
||
|
||
verify_status, verify_content = get_page(host, repo, token, page)
|
||
if verify_status == "ok" and marker in verify_content:
|
||
return True, f"條目已寫入 {page}（第 {attempt} 次嘗試）"
|
||
log("WRN", f"第 {attempt} 次寫後驗證未找到條目，準備重試")
|
||
|
||
return False, f"重試 {retries} 次仍未成功寫入 {page}"
|
||
|
||
|
||
# ------------------------------------------------------------------------------
|
||
# 週頁命名
|
||
# ------------------------------------------------------------------------------
|
||
|
||
def week_page_name(when=None):
|
||
"""依台灣時區產生週頁名稱 Worklog-yyyy-MM-W<該月第幾週>（第幾週＝ceil(日/7)）。"""
|
||
when = when or datetime.now(TAIPEI)
|
||
week = (when.day + 6) // 7
|
||
return f"Worklog-{when.year:04d}-{when.month:02d}-W{week}"
|
||
|
||
|
||
def week_page_header(page=None, when=None):
|
||
"""產生週頁首行標題（例：# 2026 年 07 月 第 4 週工作紀錄）。"""
|
||
when = when or datetime.now(TAIPEI)
|
||
week = (when.day + 6) // 7
|
||
return f"# {when.year} 年 {when.month:02d} 月 第 {week} 週工作紀錄"
|
||
|
||
|
||
# ------------------------------------------------------------------------------
|
||
# CLI
|
||
# ------------------------------------------------------------------------------
|
||
|
||
USAGE = """用法：wiki_api.py <子命令> [參數]
|
||
|
||
probe 檢查 host／repo／token／wiki API 可用性
|
||
page-name 印出當週頁面名稱
|
||
pages 列出全部頁面（title 與實際 sub_url）
|
||
show [頁面] 印出指定頁面內容（預設當週頁）
|
||
append <marker> [頁面] 自 stdin 讀取條目內容並追加（預設當週頁）
|
||
init [頁面] 若當週頁不存在則建立（僅含標題）
|
||
delete <頁面> 刪除指定頁面
|
||
|
||
環境變數：WORKLOG_HOST（必要）、WORKLOG_REPO（必要）、GITEA_TOKEN（選用，會自動 fallback）
|
||
"""
|
||
|
||
|
||
def _env():
|
||
"""讀取並檢查必要環境變數，回傳 (host, repo)；缺少時結束程式。"""
|
||
host = os.environ.get("WORKLOG_HOST", "").strip()
|
||
repo = os.environ.get("WORKLOG_REPO", "").strip()
|
||
if not host or not repo:
|
||
log("ERR", "缺少 WORKLOG_HOST 或 WORKLOG_REPO")
|
||
sys.exit(2)
|
||
return host, repo
|
||
|
||
|
||
def main(argv):
|
||
"""CLI 進入點：解析子命令並執行對應 wiki 操作。"""
|
||
if not argv or argv[0] in ("-h", "--help"):
|
||
print(USAGE)
|
||
return 0
|
||
|
||
cmd = argv[0]
|
||
|
||
if cmd == "page-name":
|
||
print(week_page_name())
|
||
return 0
|
||
|
||
host, repo = _env()
|
||
token, source = resolve_token(host, repo)
|
||
if not token:
|
||
log("ERR", f"無可用 token：{source}")
|
||
return 2
|
||
|
||
if cmd == "probe":
|
||
log("INF", f"token 來源：{source}")
|
||
code, body = _request("GET", f"https://{host}/api/v1/version", token, None)
|
||
log("INF", f"Gitea 版本查詢 HTTP {code} {body[:80]}")
|
||
status, _ = get_page(host, repo, token, week_page_name())
|
||
log("INF", f"當週頁 {week_page_name()} 狀態：{status}")
|
||
return 0
|
||
|
||
if cmd == "pages":
|
||
status, pages = list_pages(host, repo, token)
|
||
if status != "ok":
|
||
log("WRN" if status == "missing" else "ERR", f"頁面清單狀態：{status}")
|
||
return 0 if status == "missing" else 1
|
||
for item in pages:
|
||
print(f"{item.get('title')}\t{item.get('sub_url')}")
|
||
return 0
|
||
|
||
if cmd == "delete":
|
||
if len(argv) < 2:
|
||
log("ERR", "delete 需要頁面名稱")
|
||
return 2
|
||
ok, msg = delete_page(host, repo, token, argv[1])
|
||
log("INF" if ok else "ERR", msg)
|
||
return 0 if ok else 1
|
||
|
||
if cmd == "show":
|
||
page = argv[1] if len(argv) > 1 else week_page_name()
|
||
status, content = get_page(host, repo, token, page)
|
||
if status == "ok":
|
||
print(content)
|
||
return 0
|
||
log("WRN" if status == "missing" else "ERR", f"頁面 {page} 狀態：{status} {content}")
|
||
return 0 if status == "missing" else 1
|
||
|
||
if cmd == "init":
|
||
page = argv[1] if len(argv) > 1 else week_page_name()
|
||
status, _ = get_page(host, repo, token, page)
|
||
if status == "ok":
|
||
log("INF", f"頁面 {page} 已存在，不重建")
|
||
return 0
|
||
ok, msg = create_page(host, repo, token, page, week_page_header(page) + "\n", f"worklog: 初始化 {page}")
|
||
log("INF" if ok else "ERR", msg)
|
||
return 0 if ok else 1
|
||
|
||
if cmd == "append":
|
||
if len(argv) < 2:
|
||
log("ERR", "append 需要 marker 參數")
|
||
return 2
|
||
marker = argv[1]
|
||
page = argv[2] if len(argv) > 2 else week_page_name()
|
||
entry = sys.stdin.read().strip()
|
||
if not entry:
|
||
log("WRN", "條目內容為空，不寫入")
|
||
return 0
|
||
ok, msg = append_entry(host, repo, token, page, week_page_header(page), entry, marker)
|
||
log("INF" if ok else "ERR", msg)
|
||
return 0 if ok else 1
|
||
|
||
log("ERR", f"未知子命令：{cmd}")
|
||
print(USAGE)
|
||
return 2
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main(sys.argv[1:]))
|