#!/usr/bin/env python3 """package.py — 把抓取结果整理成可检索的归档：README.md + INDEX.md + tar.gz + git 提交。 产物 - README.md 归档总览：来源、统计、目录说明、版权与缺口 - INDEX.md 每组一篇索引（标题 + 本地路径 + 来源 URL） - ARCHIVE.tar.gz 本地全量快照（含 raw/ 原始响应） - git 仓库 只提交 md/ + repos/ + manifest + 索引（raw/ 不入库，避免几百 MB 的 HTML） 用法: python3 package.py [--push ] """ from __future__ import annotations import argparse import collections import datetime import re import subprocess import sys import tarfile from pathlib import Path ROOT = Path(__file__).resolve().parent GROUPS = { 'zn-help': '紫鸟帮助中心（help/docs + help-v5/docs，共 468 页）', 'zn-blog': '紫鸟跨境知识库（WordPress 博客，255 篇）', 'zn-news': '紫鸟主站资讯与博客（/news + /blog，含 SEO 文章约 3000 篇）', 'hs-api': 'HubStudio API 文档（api-docs，Markdown 直出，160 页）', 'hs-support': 'HubStudio 帮助中心（support 站，首页可见的文章）', } SOURCES = [ ('紫鸟 sitemap', 'https://www.ziniao.com/sitemap.xml'), ('紫鸟博客 sitemap', 'https://blog.ziniao.com/sitemaps.xml'), ('紫鸟帮助中心', 'https://www.ziniao.com/help/'), ('紫鸟开放平台', 'https://open.ziniao.com/'), ('紫鸟官方 Skill 仓库', 'https://github.com/ziniao-open/skills'), ('紫鸟官方 WebDriver/Skill Hub', 'https://github.com/ziniao-open/skill-hub'), ('HubStudio API 文档 sitemap', 'https://api-docs.hubstudio.cn/sitemap.xml'), ('HubStudio 帮助中心', 'https://support-orig.hubstudio.cn/'), ('HubStudio 官方 Skill 仓库', 'https://github.com/hubstudio-Max/hubstudio-skill'), ] def group_stats() -> dict: stats = {} manifest = ROOT / 'manifest.tsv' rows = [] if manifest.exists(): for ln in manifest.read_text(encoding='utf-8').splitlines(): parts = ln.split('\t') if len(parts) >= 4: rows.append(parts) for g in GROUPS: mds = list((ROOT / 'md' / g).rglob('*.md')) raws = list((ROOT / 'raw' / g).rglob('*')) if (ROOT / 'raw' / g).exists() else [] raws = [f for f in raws if f.is_file()] ok = sum(1 for r in rows if r[0] == g and r[2] == '200') fail = sum(1 for r in rows if r[0] == g and r[2] != '200') stats[g] = { 'md': len(mds), 'raw': len(raws), 'list': len(_urls(g)), 'ok': ok, 'fail': fail, 'md_bytes': sum(f.stat().st_size for f in mds), 'raw_bytes': sum(f.stat().st_size for f in raws), } return stats def _urls(group: str) -> list[str]: f = ROOT / f'urls-{group}.txt' return [x for x in f.read_text(encoding='utf-8').splitlines() if x.strip()] if f.exists() else [] def title_of(md: Path) -> str: try: for ln in md.read_text(encoding='utf-8', errors='replace').splitlines()[:40]: if ln.startswith('#'): return ln.lstrip('# ').strip() for ln in md.read_text(encoding='utf-8', errors='replace').splitlines()[:6]: if 'source:' in ln: return md.stem except Exception: # noqa: BLE001 pass return md.stem def write_readme(stats: dict) -> None: now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M') total_md = sum(s['md'] for s in stats.values()) total_raw = sum(s['raw'] for s in stats.values()) total_mb = sum(s['md_bytes'] + s['raw_bytes'] for s in stats.values()) / 2**20 lines = [ '# 紫鸟 / HubStudio 文档归档', '', f'抓取时间：**{now}**　抓取工具：`crawl.py`（本目录，源码随包提交）', '', '## 目录结构', '', '| 路径 | 说明 |', '|---|---|', '| `md//…` | 提取后的 Markdown（可直接阅读/检索，gitea 里有） |', '| `raw//…` | 原始响应（HTML 或 .md），**仅本地保留**，不入 git |', '| `repos/` | 三个官方 Skill 仓库的文件级快照（见下） |', '| `manifest.tsv` | 每页记录：group / url / http / bytes / sha256前16 / title |', '| `INDEX.md` | 分组索引：标题 + 本地路径 + 来源 URL |', '| `ARCHIVE.tar.gz` | 本地全量快照（含 raw/） |', '', '## 统计', '', '| group | 说明 | 计划 | 成功 | 失败 | md | raw | 体积 |', '|---|---|---:|---:|---:|---:|---:|---:|', ] for g, desc in GROUPS.items(): s = stats[g] lines.append(f"| `{g}` | {desc} | {s['list']} | {s['ok']} | {s['fail']} | {s['md']} | {s['raw']} | " f"{s['md_bytes']/2**20:.1f}M + {s['raw_bytes']/2**20:.1f}M |") lines += ['', f'合计：**{total_md} 篇 Markdown**、{total_raw} 份原始响应、约 **{total_mb:.0f} MB**。', '', '## 来源', ''] lines += [f'- {n}：<{u}>' for n, u in SOURCES] lines += ['', '## 官方 Skill 仓库（repos/）', '', '| 仓库 | 内容 |', '|---|---|', '| `repos/skills` | 紫鸟官方 skill 包（员工/角色/部门/账号/设备/店铺控制等），CC BY 4.0 |', '| `repos/skill-hub` | 紫鸟 WebDriver 文档与示例（applyAuth/getBrowserList/startBrowser…） |', '| `repos/hubstudio-skill` | HubStudio 官方 skill（含 ADB 连接、OpenClaw 教程、commands 清单） |', '', '## 已知缺口（重要）', '', '- **HubStudio 帮助中心（support 站）是 Baklib 托管的 SPA**：文章页无服务端内链、站点无 sitemap，', ' 只能取到首页可见的文章（约 20 篇）。要抓全需登录态或渲染 JS，本归档不含。', '- 紫鸟主站 `/blog/*` 与 WordPress 博客是两套内容，均按原样归档，可能有重叠主题。', '- 页面内的图片/附件未下载（只存正文文本）；需要时按 `raw/` 里的原始 URL 再取。', '- 部分页面为 404/登录后可见，manifest 里 http 非 200 的行即失败项。', '', '## 版权与使用', '', '- 文档内容版权归各自厂商（紫鸟：福建紫讯信息科技；HubStudio：其运营方）所有，**本归档仅供内部查阅参考**，', ' 不改变原授权；对外分发前请先确认授权。', '- `repos/skills`、`repos/skill-hub`、`repos/hubstudio-skill` 为官方公开仓库快照，' '紫鸟 skill 包声明为 **CC BY 4.0**，使用时请保留署名。', '', '## 复现', '', '```bash', 'python3 crawl.py --list # 看各来源规模', 'python3 crawl.py # 全量抓取（幂等，可续抓）', 'python3 mirror_repos.py # 快照官方仓库', 'python3 package.py --push ', '```', ''] (ROOT / 'README.md').write_text('\n'.join(lines), encoding='utf-8') def write_index() -> None: out = ['# 索引', ''] for g, desc in GROUPS.items(): mds = sorted((ROOT / 'md' / g).rglob('*.md')) out += [f'## {g} — {desc}', '', f'共 {len(mds)} 篇。', '', '| 标题 | 本地路径 |', '|---|---|'] for md in mds: rel = md.relative_to(ROOT) t = title_of(md).replace('|', '/')[:90] out.append(f'| {t or md.stem} | `{rel}` |') out.append('') (ROOT / 'INDEX.md').write_text('\n'.join(out), encoding='utf-8') def make_tar() -> None: tar = ROOT / 'ARCHIVE.tar.gz' with tarfile.open(tar, 'w:gz') as tf: for sub in ('md', 'raw', 'repos'): p = ROOT / sub if p.exists(): tf.add(p, arcname=sub) for f in ('manifest.tsv', 'INDEX.md', 'README.md', 'crawl.py', 'mirror_repos.py', 'package.py'): if (ROOT / f).exists(): tf.add(ROOT / f, arcname=f) print(f' ARCHIVE.tar.gz: {tar.stat().st_size / 2**20:.1f} MB') def git_commit_push(remote: str | None, stats: dict) -> None: """gitea 侧只提交 md/ + repos/ + 元数据（raw/ 不入库）。""" (ROOT / '.gitignore').write_text('raw/\nARCHIVE.tar.gz\ncrawl.log\n*.bak\n__pycache__/\n') cmds = [['git', 'init', '-q'], ['git', 'add', '-A'], ['git', '-c', 'user.name=admin9webs', '-c', 'user.email=admin9webs@9webs.site', 'commit', '-q', '-m', f'archive: 紫鸟/HubStudio 文档快照（{sum(s["md"] for s in stats.values())} 篇 Markdown）\n\n' '来源：紫鸟帮助中心/博客/主站资讯 + HubStudio API 文档/帮助中心 + 三个官方 skill 仓库。\n' '只入库 Markdown 与元数据；原始 HTML 与 tarball 留在本机。']] for c in cmds: r = subprocess.run(c, cwd=ROOT, capture_output=True, text=True) if r.returncode != 0 and 'nothing to commit' not in (r.stdout + r.stderr): print(' git:', ' '.join(c[1:3]), '->', (r.stdout + r.stderr).strip()[:200]) if remote: subprocess.run(['git', 'remote', 'remove', 'origin'], cwd=ROOT, capture_output=True) subprocess.run(['git', 'remote', 'add', 'origin', remote], cwd=ROOT, check=False) r = subprocess.run(['git', 'push', '-u', 'origin', 'HEAD:main'], cwd=ROOT, capture_output=True, text=True) out = re.sub(r'://[^@\s]+@', '://***@', (r.stdout + r.stderr).strip()) print(' push:', out[-200:] if r.returncode else 'OK') def main() -> int: ap = argparse.ArgumentParser() ap.add_argument('--push', default=None, help='gitea remote（含凭据），推送归档') ap.add_argument('--no-tar', action='store_true') args = ap.parse_args() stats = group_stats() write_readme(stats) write_index() print(' README.md / INDEX.md 已生成') for g, s in stats.items(): print(f" {g:11s} md={s['md']:4d} ok={s['ok']:4d} fail={s['fail']:3d} " f"({s['md_bytes']/2**20:.1f}M md)") if not args.no_tar: make_tar() git_commit_push(args.push, stats) return 0 if __name__ == '__main__': sys.exit(main())