wuyang
310 lines
17 KiB
Python
310 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""Create a source-backed PawPlay weekly report without an LLM request."""
|
|
import argparse
|
|
import calendar
|
|
import datetime as dt
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from zoneinfo import ZoneInfo
|
|
|
|
TZ = ZoneInfo('Asia/Shanghai')
|
|
|
|
|
|
def read_json(path, default=None):
|
|
try:
|
|
return json.loads(Path(path).read_text())
|
|
except FileNotFoundError:
|
|
return default
|
|
|
|
|
|
def save_json(path, value):
|
|
path = Path(path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with tempfile.NamedTemporaryFile('w', dir=path.parent, delete=False, encoding='utf-8') as f:
|
|
json.dump(value, f, ensure_ascii=False, indent=2)
|
|
f.write('\n')
|
|
staging = f.name
|
|
os.replace(staging, path)
|
|
|
|
|
|
def subscription_allocation(budget, start, end):
|
|
"""Prorate calendar days; token counts are never a second subscription bill."""
|
|
price = budget.get('subscription_usd')
|
|
share = budget.get('project_allocation_fraction')
|
|
period = budget.get('billing_period')
|
|
if price is None or share is None or period not in ('monthly', 'weekly'):
|
|
return None
|
|
if not isinstance(start, dt.datetime):
|
|
start = dt.datetime.combine(start, dt.time(), TZ)
|
|
if not isinstance(end, dt.datetime):
|
|
end = dt.datetime.combine(end, dt.time(), TZ)
|
|
total = 0.0
|
|
cursor = start.astimezone(TZ)
|
|
end = end.astimezone(TZ)
|
|
while cursor < end:
|
|
day = cursor.date()
|
|
divisor = calendar.monthrange(day.year, day.month)[1] if period == 'monthly' else 7
|
|
next_midnight = dt.datetime.combine(day + dt.timedelta(days=1), dt.time(), TZ)
|
|
step_end = min(next_midnight, end)
|
|
total += float(price) * float(share) / divisor * ((step_end - cursor).total_seconds() / 86400)
|
|
cursor = step_end
|
|
return round(total, 2)
|
|
|
|
|
|
def session_metadata(path):
|
|
with path.open() as f:
|
|
for _ in range(4):
|
|
line = f.readline()
|
|
if not line:
|
|
break
|
|
try:
|
|
value = json.loads(line)
|
|
except ValueError:
|
|
continue
|
|
if value.get('type') == 'session_meta':
|
|
return value.get('payload', {})
|
|
return {}
|
|
|
|
|
|
def codex_usage(session_root, thread_id, start, end):
|
|
"""Use cumulative counter deltas, scoped to the chat and its descendants."""
|
|
if not session_root or not thread_id:
|
|
return {'available': False, 'reason': '项目会话来源未配置'}
|
|
candidates = []
|
|
for path in Path(session_root).rglob('*.jsonl'):
|
|
meta = session_metadata(path)
|
|
candidates.append((path, meta))
|
|
allowed = {thread_id}
|
|
changed = True
|
|
while changed:
|
|
changed = False
|
|
for _, meta in candidates:
|
|
source = meta.get('source', {})
|
|
spawn = source.get('subagent', {}).get('thread_spawn', {}) if isinstance(source, dict) else {}
|
|
if spawn.get('parent_thread_id') in allowed and meta.get('id') not in allowed:
|
|
allowed.add(meta['id'])
|
|
changed = True
|
|
totals = {k: 0 for k in ('input_tokens', 'cached_input_tokens', 'output_tokens', 'reasoning_output_tokens', 'total_tokens')}
|
|
sessions = 0
|
|
for path, meta in candidates:
|
|
if meta.get('id') not in allowed:
|
|
continue
|
|
sessions += 1
|
|
previous = dict.fromkeys(totals, 0)
|
|
with path.open() as handle:
|
|
for line in handle:
|
|
try:
|
|
event = json.loads(line)
|
|
payload = event.get('payload', {})
|
|
if event.get('type') != 'event_msg' or payload.get('type') != 'token_count':
|
|
continue
|
|
current = (payload.get('info') or {}).get('total_token_usage')
|
|
if not current:
|
|
continue
|
|
stamp = dt.datetime.fromisoformat(event['timestamp'].replace('Z', '+00:00'))
|
|
if start <= stamp < end:
|
|
for key in totals:
|
|
totals[key] += max(0, int(current.get(key, 0)) - previous[key])
|
|
previous = {k: int(current.get(k, 0)) for k in totals}
|
|
except (ValueError, KeyError, TypeError):
|
|
continue
|
|
return {'available': sessions > 0, 'session_count': sessions, **totals,
|
|
'uncached_input_tokens': max(0, totals['input_tokens'] - totals['cached_input_tokens']),
|
|
'measurement_note': '会话日志计数含重复处理的上下文和缓存；不是唯一文字数、周额度百分比或API账单。推理token包含于输出计数，不重复相加。'}
|
|
|
|
|
|
def scheduled_codex_usage(directory, start, end):
|
|
"""Read bounded cycle summaries separately from the interactive thread."""
|
|
if not directory or not Path(directory).is_dir():
|
|
return {'available': False, 'reason': '定时运营汇总目录未配置或不可用'}
|
|
totals = {k: 0 for k in ('input_tokens', 'cached_input_tokens', 'output_tokens', 'reasoning_output_tokens')}
|
|
api_totals = dict.fromkeys(totals, 0)
|
|
runs = failed = 0
|
|
categories = {}
|
|
api_runs = api_uncertain = api_usage_events = 0
|
|
for path in Path(directory).glob('*.summary.json'):
|
|
try:
|
|
record = json.loads(path.read_text())
|
|
stamp = record.get('started_at')
|
|
when = dt.datetime.fromisoformat(stamp.replace('Z', '+00:00')) if stamp else dt.datetime.strptime(record['id'][:16], '%Y%m%dT%H%M%SZ').replace(tzinfo=dt.timezone.utc)
|
|
if when.tzinfo is None or not start <= when < end:
|
|
continue
|
|
runs += 1
|
|
failed += record.get('status') != 'complete'
|
|
if record.get('auth') == 'api_key':
|
|
api_runs += record.get('api_fallback_used') is True
|
|
api_uncertain += record.get('api_fallback_used') is None
|
|
for usage in record.get('usage', []):
|
|
api_usage_events += 1
|
|
for k in api_totals:
|
|
api_totals[k] += max(0, int(usage.get(k, 0)))
|
|
continue
|
|
category = record.get('category') if record.get('category') in ('content', 'engineering', 'research', 'community', 'management') else 'unattributed'
|
|
lane = categories.setdefault(category, {'runs': 0, 'elapsed_seconds': 0, 'elapsed_missing': 0, **dict.fromkeys(totals, 0)})
|
|
lane['runs'] += 1
|
|
if isinstance(record.get('elapsed_seconds'), (int, float)) and record['elapsed_seconds'] >= 0: lane['elapsed_seconds'] += record['elapsed_seconds']
|
|
else: lane['elapsed_missing'] += 1
|
|
for usage in record.get('usage', []):
|
|
for k in totals:
|
|
amount = max(0, int(usage.get(k, 0)))
|
|
totals[k] += amount
|
|
lane[k] += amount
|
|
except (ValueError, OSError, KeyError, TypeError, AttributeError):
|
|
continue
|
|
return {'available': True, 'runs': runs, 'failed_runs': failed, 'api_runs_separate': api_runs, 'api_runs_uncertain': api_uncertain, 'api_observed_usage': api_totals if api_usage_events else None, 'by_category': categories, **totals,
|
|
'measurement_note': '来自独立定时会话的运行汇总，与交互会话分开统计；API 备用调用不计入订阅 token，实际成本单独核算。失败或中断可能没有完整用量。'}
|
|
|
|
|
|
def harness_run_usage(path, start, end):
|
|
if not Path(path).exists():
|
|
return {'available': False, 'reason': '后续 Harness 调用账本尚未创建'}
|
|
result = {'available': True, 'runs': 0, 'input_tokens': 0, 'output_tokens': 0, 'cash_settlement_usd': None}
|
|
for line in Path(path).read_text().splitlines():
|
|
try:
|
|
row = json.loads(line)
|
|
when = dt.datetime.fromisoformat(row['at_utc'].replace('Z', '+00:00'))
|
|
if when.tzinfo is None or not start <= when < end:
|
|
continue
|
|
result['runs'] += 1
|
|
for usage in row.get('usage') or []:
|
|
result['input_tokens'] += max(0, int(usage.get('inputTokens', 0)))
|
|
result['output_tokens'] += max(0, int(usage.get('outputTokens', 0)))
|
|
except (ValueError, KeyError, TypeError, AttributeError):
|
|
continue
|
|
result['measurement_note'] = '只统计自动账本创建后的调用；实际结算未知，不与历史快照重复计费。'
|
|
return result
|
|
|
|
|
|
def build_report(data_dir, snapshot=False, now=None):
|
|
now = now or dt.datetime.now(TZ)
|
|
now = now.astimezone(TZ)
|
|
cutoff = now.replace(hour=20, minute=0, second=0, microsecond=0) - dt.timedelta(days=(now.weekday() - 4) % 7)
|
|
if cutoff > now:
|
|
cutoff -= dt.timedelta(days=7)
|
|
begin_stamp = cutoff if snapshot else cutoff - dt.timedelta(days=7)
|
|
end_stamp = now if snapshot else cutoff
|
|
start, end = begin_stamp.date(), end_stamp.date()
|
|
config = read_json(data_dir / 'config.json', {})
|
|
collector = read_json(data_dir / 'collector.json', {})
|
|
ledger = read_json(data_dir / 'ledger.json', {'entries': []})
|
|
content = read_json(data_dir / 'content.json', {'items': []})
|
|
tasks = read_json(data_dir / 'tasks.json', {'items': []})
|
|
queue = read_json(data_dir / 'workqueue.json', None)
|
|
findings = read_json(data_dir / 'findings.json', {})
|
|
api = read_json(data_dir / 'deepseek-usage.json', {})
|
|
entries = []
|
|
for entry in ledger.get('entries', []):
|
|
if entry.get('project') not in config.get('projects', ['pawplay']):
|
|
continue
|
|
stamp = entry.get('date', '')
|
|
try:
|
|
value = dt.datetime.fromisoformat(stamp.replace('Z', '+00:00'))
|
|
if value.tzinfo is None:
|
|
value = value.replace(tzinfo=TZ)
|
|
if begin_stamp <= value < end_stamp:
|
|
entries.append(entry)
|
|
except ValueError:
|
|
pass
|
|
published = [i for i in content.get('items', []) if i.get('status') == 'published' and i.get('bvid')]
|
|
costs = [e for e in entries if e.get('direction') == 'expense' and e.get('currency') == 'USD' and e.get('basis') == 'cash' and e.get('amount') is not None]
|
|
incomes = [e for e in entries if e.get('direction') == 'income' and e.get('currency') == 'CNY' and e.get('basis') == 'settled' and e.get('amount') is not None]
|
|
allocated = subscription_allocation(config.get('budget', {}), begin_stamp, end_stamp)
|
|
usage = codex_usage(collector.get('codex_sessions_dir'), collector.get('thread_id'), begin_stamp, end_stamp)
|
|
def observed_this_period(stamp):
|
|
try:
|
|
value = dt.datetime.fromisoformat(stamp.replace('Z', '+00:00'))
|
|
if value.tzinfo is None:
|
|
value = value.replace(tzinfo=TZ)
|
|
return begin_stamp <= value < end_stamp
|
|
except (AttributeError, ValueError):
|
|
return False
|
|
highlights = []
|
|
for item in findings.get('highlights', []):
|
|
stamp = item.get('date') if isinstance(item, dict) else findings.get('checked_at')
|
|
if observed_this_period(stamp):
|
|
highlights.append(item.get('text', '') if isinstance(item, dict) else item)
|
|
risks = list(findings.get('risks', []))
|
|
if not observed_this_period(findings.get('checked_at')):
|
|
risks.append('基础验收记录未在本周期更新，历史成果未重复计作本周完成。')
|
|
current = next((i for i in tasks.get('items', []) if i.get('status') == 'in_progress'), None)
|
|
next_actions = [i['title'] for i in tasks.get('items', []) if i.get('status') in ('todo', 'in_progress')][:8]
|
|
# A task queue is a current snapshot; it does not reconstruct the past reporting period.
|
|
if isinstance(queue, dict) and queue.get('schema') == 1 and isinstance(queue.get('tasks'), list):
|
|
safe = [i for i in queue['tasks'] if isinstance(i, dict) and i.get('status') in ('pending', 'running', 'blocked') and isinstance(i.get('title'), str) and isinstance(i.get('id'), str)]
|
|
safe.sort(key=lambda i: (0 if i['status'] == 'running' else 1 if i['status'] == 'pending' else 2, i.get('updated_at') or ''))
|
|
next_actions = [('核查新互动' if i['id'].startswith('interaction-') else i['title'][:100]) + ('（已阻塞）' if i['status'] == 'blocked' else '') for i in safe[:8]]
|
|
current = next(({'title': '核查新互动' if i['id'].startswith('interaction-') else i['title'][:100]} for i in safe if i['status'] == 'running'), None)
|
|
|
|
identifier = end.strftime('%G-W%V') + ('-snapshot-' + now.strftime('%Y%m%dT%H%M%S') if snapshot else '')
|
|
report = {
|
|
'id': identifier,
|
|
'title': '账号运营' + ('阶段快照' if snapshot else '周报') + ' ' + end.isoformat(),
|
|
'period_start': begin_stamp.isoformat(), 'period_end': end_stamp.isoformat(),
|
|
'published_at': now.isoformat(), 'partial_period': snapshot, 'kind': 'snapshot' if snapshot else 'weekly',
|
|
'summary': ('当前推进：' + current['title']) if current else '根据已归档的项目记录汇总；缺失数据保留待核算。',
|
|
'highlights': highlights,
|
|
'risks': risks,
|
|
'next_actions': next_actions,
|
|
'decisions': findings.get('decisions', []),
|
|
'metrics': {
|
|
'published_project_videos': len(published),
|
|
'project_video_views': 0 if not published else None,
|
|
'project_game_sessions': None,
|
|
'codex': usage,
|
|
'scheduled_codex': scheduled_codex_usage(collector.get('ops_cycle_summaries_dir'), begin_stamp, end_stamp),
|
|
'deepseek': api or {'status': '未采集'},
|
|
'harness_new_runs': harness_run_usage(data_dir / 'harness-runs.jsonl', begin_stamp, end_stamp),
|
|
'measurement_notes': ['只统计运营账本登记稿件，不计入账户其他历史内容。', '当前会话包含PawPlay、运营基础设施和虚拟人设研究，token仅能按该共享会话汇总，未精确归因到单个方向。定时任务作为独立会话分别统计。', '游戏离线运行且未接入统计，不能推导活跃、留存或转化率。', '已发布稿件存在时，播放数据需接入稿件快照采集，当前不填造数值。']
|
|
},
|
|
'costs': {
|
|
'cash_usd': round(sum(e['amount'] for e in costs), 2) if config.get('cash_tracking_complete') else None,
|
|
'confirmed_recorded_cash_usd': round(sum(e['amount'] for e in costs), 2),
|
|
'allocated_usd': allocated, 'estimated_usd': None,
|
|
'electricity_kwh': None, 'electricity_cny': None,
|
|
'notes': ['固定套餐按计费周期的日历天数分摊；项目占比待确认时不报项目分摊金额。', '不按token再次给套餐内使用计费。DeepSeek/API如另计费，需用用量及实际费率/账单入账。', '电费需功率积分和电价，NAS/VPS及硬件折旧需确认项目分摊口径；不填虚构零成本。']
|
|
},
|
|
'revenue': {'settled_cny': round(sum(e['amount'] for e in incomes), 2) if incomes else None, 'estimated_cny': None,
|
|
'notes': ['播放量不是结算收入，自有产品推广稿件是否纳入激励需单独验证。', '未结算或未归因到PawPlay的收益不算项目收入。']},
|
|
'sources': {'findings_checked_at': findings.get('checked_at'), 'ledger_entries_in_period': len(entries), 'content_ledger': 'content.json', 'next_actions_source': 'workqueue.json：报告生成时的任务快照' if isinstance(queue, dict) and queue.get('schema') == 1 and isinstance(queue.get('tasks'), list) else 'tasks.json：静态任务清单', 'usage_source': '本机会话计数（只保存汇总，不发布对话）'},
|
|
}
|
|
repo = collector.get('repository')
|
|
if repo:
|
|
r = subprocess.run(['git', '-C', repo, 'rev-parse', '--short', 'HEAD'], capture_output=True, text=True)
|
|
report['sources']['repository_commit'] = r.stdout.strip() if r.returncode == 0 else None
|
|
target = data_dir / 'reports' / (identifier + '.json')
|
|
# A finalized weekly report is immutable. Manual corrections require a new revision.
|
|
if target.exists():
|
|
return target, False
|
|
save_json(target, report)
|
|
return target, True
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--data-dir', required=True, type=Path)
|
|
parser.add_argument('--snapshot', action='store_true')
|
|
parser.add_argument('--notify', action='store_true')
|
|
args = parser.parse_args()
|
|
target, created = build_report(args.data_dir, args.snapshot)
|
|
print(json.dumps({'created': created, 'report': target.name}, ensure_ascii=False))
|
|
collector = read_json(args.data_dir / 'collector.json', {})
|
|
if created and collector.get('backup_dir'):
|
|
backup = Path(collector['backup_dir']) / target.stem
|
|
backup.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
for name in ('config.json', 'ledger.json', 'content.json', 'tasks.json', 'findings.json', 'deepseek-usage.json'):
|
|
if (args.data_dir / name).exists():
|
|
shutil.copy2(args.data_dir / name, backup / name)
|
|
shutil.copy2(target, backup / target.name)
|
|
if created and args.notify:
|
|
script = collector.get('bark_script')
|
|
if script:
|
|
subprocess.run(['python3', script, 'PawPlay 周报已归档，可登录运营网站查看预算、收益及下周工作。', '--title', 'PawPlay：每周运营报告', '--url', collector.get('portal_url', '')], check=True)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|