doc/scripts/worklog/wiki_api.py

#!/usr/bin/env python3
# ==============================================================================
# 用途：Gitea Wiki 讀寫工具（worklog 專用）。提供 token 解析、頁面讀取、
#       建立、append 追加（read-modify-write + 寫後驗證重試），供 worklog.sh
#       與 /jsc-doc: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:]))