fnvtk/karuo-ai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🔄 卡若AI 同步 2026-03-12 20:45 | 更新：水桥平台对接、运营中枢工作台 | 排除 >20MB: 11 个

Browse Source
This commit is contained in:
karuo
2026-03-12 20:45:20 +08:00
parent 920372abfb
commit 3d0f6c3cf0
5 changed files with 105 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
02_卡人（水）/水桥_平台对接/飞书管理/脚本/.feishu_tokens.json
View File

@@ -1,6 +1,6 @@
{
"access_token": "u-fC.wuuRp91Ha5dCmygjeK5lh1CzxghOPiMGaUwk0261U",
"refresh_token": "ur-c49nCzxYZcNEgDW6CQEb0Rlh1AHxghqXqwGaYNk0270J",
"access_token": "u-e9uHxerhZ7VrnwkCFC9Yvalh3Ix1ghWXpgGaZMk0260Y",
"refresh_token": "ur-dIaKx4ssV4SU.karG6lt9ulh1C11ghopOgGaYx00261E",
"name": "飞书用户",
"auth_time": "2026-03-12T12:18:56.479463"
"auth_time": "2026-03-12T20:33:34.954705"
}

206
02_卡人（水）/水桥_平台对接/飞书管理/脚本/auto_log.py
View File

@@ -192,129 +192,68 @@ def get_today_tasks():
return date_str, tasks
def _tb(content, bold=False):
"""构建纯文本 block（兼容性最强，不带 text_color/align）"""
elem = {'text_run': {'content': content}}
if bold:
elem['text_run']['text_element_style'] = {'bold': True}
return {'block_type': 2, 'text': {'elements': [elem], 'style': {}}}
def _todo(content):
"""待办块（不带 align，减少 field validation failed 风险）"""
return {'block_type': 17, 'todo': {'elements': [{'text_run': {'content': content}}], 'style': {'done': False}}}
def build_blocks(date_str, tasks):
"""构建飞书文档块（美观版 TNTWF 格式，不含易报错的 callout/text_color/align）"""
QUADRANT_ICONS = {
"重要紧急": "🔴 重要紧急",
"重要不紧急": "🟡 重要不紧急",
"不重要紧急": "🔵 不重要紧急",
"不重要不紧急":"⚪ 不重要不紧急",
}
quadrant_order = ["重要紧急", "重要不紧急", "不重要紧急", "不重要不紧急"]
"""构建飞书文档块（倒序：新日期在上）；沿用 2 月旧版 TNTWF 结构"""
blocks = [
_tb(f'📅 {date_str}', bold=True),
_tb('▶ 执行'),
_tb(''),
{'block_type': 6, 'heading4': {'elements': [{'text_run': {'content': f'{date_str} '}}], 'style': {'align': 1}}},
{'block_type': 2, 'text': {'elements': [{'text_run': {'content': '[执行]', 'text_element_style': {'bold': True}}}], 'style': {}}}
]
quadrant_colors = {"重要紧急": 5, "重要不紧急": 3, "不重要紧急": 6, "不重要不紧急": 4}
quadrant_order = ["重要紧急", "重要不紧急", "不重要紧急", "不重要不紧急"]
for quadrant in quadrant_order:
q_tasks = [t for t in tasks if t.get('quadrant') == quadrant]
if not q_tasks:
continue
blocks.append(_tb(f' {QUADRANT_ICONS.get(quadrant, quadrant)}', bold=True))
blocks.append(_tb(' ' + '─' * 32))
blocks.append({'block_type': 2, 'text': {'elements': [{'text_run': {'content': f'[{quadrant}]',
'text_element_style': {'bold': True, 'text_color': quadrant_colors[quadrant]}}}], 'style': {'align': 1}}})
for task in q_tasks:
events = ' · '.join(task.get('events', []))
blocks.append(_todo(f"{task.get('person', '')}（{events}）"))
events = "、".join(task['events'])
blocks.append({'block_type': 17, 'todo': {'elements': [{'text_run': {'content': f"{task['person']}（{events}）"}}],
'style': {'done': False, 'align': 1}}})
blocks.append({'block_type': 2, 'text': {'elements': [{'text_run': {'content': '{'}}], 'style': {}}})
# TNTWF格式：仅 W(工作) F(反馈) 有复选框，T/N/T 为纯文本
labels = [
('T', 't_targets', '目标'),
('N', 'n_process', '过程'),
('T', 't_thoughts', '思考'),
('W', 'w_work', '工作'),
('F', 'f_feedback', '反馈'),
('T', 't_targets', '目标', False),
('N', 'n_process', '过程', False),
('T', 't_thoughts', '思考', False),
('W', 'w_work', '工作', True),
('F', 'f_feedback', '反馈', True)
]
for label, key, name in labels:
for label, key, name, use_todo in labels:
items = task.get(key, [])
if not items:
continue
blocks.append(_tb(f' {label} {name}', bold=True))
for item in items:
if label in ('W', 'F'):
blocks.append(_todo(f' {item}'))
else:
blocks.append(_tb(f' · {item}'))
blocks.append(_tb(''))
blocks.append(_tb(''))
if items:
blocks.append({'block_type': 2, 'text': {'elements': [{'text_run': {'content': f'{label} ({name})', 'text_element_style': {'bold': True}}}], 'style': {}}})
for item in items:
if use_todo:
blocks.append({'block_type': 17, 'todo': {'elements': [{'text_run': {'content': item}}], 'style': {'done': False}}})
else:
blocks.append({'block_type': 2, 'text': {'elements': [{'text_run': {'content': item}}], 'style': {}}})
blocks.append({'block_type': 2, 'text': {'elements': [{'text_run': {'content': '}'}}], 'style': {}}})
blocks.append({'block_type': 2, 'text': {'elements': [{'text_run': {'content': ''}}], 'style': {}}})
return blocks
def _tb_s(content):
"""极简文本块（fallback 专用）"""
def _text_block_simple(content):
"""极简文本块，兼容 field validation 严格校验"""
return {'block_type': 2, 'text': {'elements': [{'text_run': {'content': content}}], 'style': {}}}
def _build_blocks_simple(date_str, tasks):
"""极简块（纯文本美观版），field validation failed 时自动回退"""
QUAD_ICONS = {
"重要紧急": "🔴 重要紧急",
"重要不紧急": "🟡 重要不紧急",
"不重要紧急": "🔵 不重要紧急",
"不重要不紧急": "⚪ 不重要不紧急",
}
quadrant_order = ["重要紧急", "重要不紧急", "不重要紧急", "不重要不紧急"]
LINE = '─' * 34
blocks = [
_tb_s(f'┌{'─' * 36}┐'),
_tb_s(f'│ 📅 {date_str} ▶ 执行' + ' ' * max(0, 30 - len(date_str)) + '│'),
_tb_s(f'└{'─' * 36}┘'),
_tb_s(''),
]
for quadrant in quadrant_order:
q_tasks = [t for t in tasks if t.get('quadrant') == quadrant]
if not q_tasks:
continue
blocks.append(_tb_s(f' {QUAD_ICONS.get(quadrant, quadrant)}'))
blocks.append(_tb_s(f' {LINE}'))
for task in q_tasks:
events = ' · '.join(task.get('events', []))
blocks.append(_tb_s(f' ☑ {task.get("person", "")}（{events}）'))
blocks.append(_tb_s(f' ┌{LINE}'))
label_map = [
('T', 't_targets', '目标'),
('N', 'n_process', '过程'),
('T', 't_thoughts', '思考'),
('W', 'w_work', '工作'),
('F', 'f_feedback', '反馈'),
]
for label, key, name in label_map:
items = task.get(key, [])
if not items:
continue
blocks.append(_tb_s(f' │ 【{label}】{name}'))
for item in items:
prefix = ' │ □ ' if label in ('W', 'F') else ' │ · '
blocks.append(_tb_s(f'{prefix}{item}'))
blocks.append(_tb_s(f' └{LINE}'))
blocks.append(_tb_s(''))
blocks.append(_tb_s(''))
"""极简块（仅纯文本），用于 field validation failed 或批量写入降级"""
blocks = [_text_block_simple(f'{date_str} '), _text_block_simple('[执行]')]
for task in tasks:
events = '、'.join(task.get('events', []))
blocks.append(_text_block_simple(f"{task.get('person', '')}（{events}）"))
for key in ('t_targets', 'n_process', 't_thoughts', 'w_work', 'f_feedback'):
for item in task.get(key, []):
blocks.append(_text_block_simple(f" {item}"))
return blocks
@@ -569,22 +508,47 @@ def write_log(token, date_str=None, tasks=None, wiki_token=None, overwrite=False
insert_index = i + 1
break
# 写入（倒序：新日期在上）；分批写入（飞书单次上限 50 块）
content_blocks = _build_blocks_simple(date_str, tasks)
BATCH_SIZE = 48 # 安全值：低于 50
offset = 0
for i in range(0, len(content_blocks), BATCH_SIZE):
batch = content_blocks[i:i + BATCH_SIZE]
payload = {'children': batch, 'index': insert_index + offset}
r = requests.post(f"https://open.feishu.cn/open-apis/docx/v1/documents/{doc_id}/blocks/{doc_id}/children",
headers=headers, json=payload, timeout=30)
rj = r.json()
if rj.get('code') != 0:
print(f"❌ 写入失败(批次{i//BATCH_SIZE+1}): code={rj.get('code')} msg={rj.get('msg')}")
return False
offset += len(batch)
print(f"✅ {date_str} 日志写入成功 -> {doc_title}（共 {len(content_blocks)} 块，{(len(content_blocks)-1)//BATCH_SIZE+1} 批次）")
return True
# 写入（倒序：新日期在上）；优先使用旧版 TNTWF 结构，必要时降级为极简文本
BATCH_SIZE = 48 # 飞书单次上限 50，取安全值
def _post_batches(blocks, label):
"""按批次写入 blocks，返回 (ok, last_error)"""
offset = 0
last_err = None
for i in range(0, len(blocks), BATCH_SIZE):
batch = blocks[i:i + BATCH_SIZE]
payload = {'children': batch, 'index': insert_index + offset}
r = requests.post(
f"https://open.feishu.cn/open-apis/docx/v1/documents/{doc_id}/blocks/{doc_id}/children",
headers=headers, json=payload, timeout=30
)
rj = r.json()
if rj.get('code') != 0:
err = f"{label} 批次{i//BATCH_SIZE+1} 失败: code={rj.get('code')} msg={rj.get('msg')}"
print(f"❌ 写入失败: {err}")
return False, err
offset += len(batch)
return True, None
# 1）先用旧版美观 TNTWF 结构
rich_blocks = build_blocks(date_str, tasks)
ok, err = _post_batches(rich_blocks, "TNTWF")
if ok:
print(f"✅ {date_str} 日志写入成功 -> {doc_title}（旧版 TNTWF 结构，共 {len(rich_blocks)} 块）")
return True
# 2）若旧版结构仍触发 field validation 或其他错误，降级为极简文本
if err and "field validation failed" in err.lower():
simple_blocks = _build_blocks_simple(date_str, tasks)
ok2, err2 = _post_batches(simple_blocks, "极简文本")
if ok2:
print(f"✅ {date_str} 日志写入成功 -> {doc_title}（极简文本降级，共 {len(simple_blocks)} 块）")
return True
print(f"❌ 极简文本写入仍失败: {err2 or 'unknown error'}")
return False
# 不是 field validation 的错误，直接失败返回
return False
def open_result(wiki_token=None):
"""打开飞书查看结果"""

80
02_卡人（水）/水桥_平台对接/飞书管理/脚本/write_today_0321_custom.py
View File

@@ -12,86 +12,36 @@ from auto_log import get_token_silent, write_log, open_result, resolve_wiki_toke
def build_tasks_today():
"""今日：200视频/日、工具研发10～30切片、售内容产出、按年度目标百分比（日期以中国时间为准）"""
"""今日：只聚焦一个核心目标——每天剪辑并分发 200 条视频（日期以中国时间为准）"""
date_str = get_today_date_str()
# 简单进度估算：按本周 3 月 10～12 日整体推进情况，粗略给出完成度区间
percent_text = "本周目标完成度预估：≈30%（工具与流程打底阶段）"
return [
{
"person": "远志（玩值）",
"events": ["200视频分发", "切片工具", "售内容与多平台推送"],
"events": ["200视频分发", "剪辑+分发一体化"],
"quadrant": "重要紧急",
"t_targets": [
"目标：每天 200 视频，工具分发到各平台",
"工具研发：每天切 10-30 个视频的切片工具",
"售方面：内容产出 + 多平台推送统一化，按年度目标 % 推进",
"核心目标：每天稳定剪辑并分发 200 条视频到全网（抖音、快手、视频号、B站等）",
"把「剪辑 → 上架 → 分发」做成一条稳定的流水线，而不是零散操作",
],
"n_process": [
"售方面还需优化迭代：优酷等各平台推送接口/API 未开发完；各公众号推送方式未写完；需把可推送的所有平台统一做出来",
"对照 3 月 10～12 日的目标，梳理本周已完成的：素材池、剪辑 SOP、分发清单等基础工作",
percent_text,
],
"t_thoughts": [
"售 = 内容产出 + 分发；多平台（优酷、公众号等）接口与推送方式补齐后，才能规模化",
"先把 200 条/日做「稳定」，再考虑扩量；优先打通从素材到上架的关键 3～5 个动作。",
],
"w_work": [
"工具分发 200 视频/日 → 各平台",
"切片工具：10-30 条/日",
"售：推到优酷等各平台的接口与 API 开发",
"售：推送到各个公众号的推送方式开发",
"统一做出所有可推送平台（优酷、公众号等）的对接与推送",
"售内容优化与迭代",
"按业务与年度目标 % 追踪",
"今天：至少完成一条完整链路的压测（从原始视频到多平台同时上线）",
"梳理「200 条/日」需要的最小人力/工具配置，并写成简洁清单",
"盘点目前可自动化的动作（批量裁切、模板套用、标题生成、分发脚本等）",
],
"f_feedback": [
"200 视频/日 进行中",
"优酷等平台接口/API 未完成",
"公众号推送方式 未完成",
"多平台统一推送 待开发",
"售优化迭代 进行中",
],
},
{
"person": "李永平",
"events": ["一场创业实验", "yongpxu-soul 分支"],
"quadrant": "重要紧急",
"t_targets": [
"永平交接「一场创业实验」+ yongpxu-soul 分支同步",
],
"n_process": [
"2/26 永平交接已启动；分支与开发进度跟进",
],
"t_thoughts": ["保持沟通，确保交接顺畅"],
"w_work": [
"一场创业实验 网站/小程序进度跟进",
"yongpxu-soul 分支同步与联调",
],
"f_feedback": [
"交接与分支 进行中",
],
},
{
"person": "卡若",
"events": ["年度目标百分比", "后台数据", "Token 过期处理"],
"quadrant": "重要不紧急",
"t_targets": [
"按业务与年度目标百分比追踪（以 2026 年整体目标为基准）",
"本月目标约 12%，距最终目标差 88%",
"后台数据链接、Token 过期直接命令处理",
],
"n_process": [
"后台数据：神射手 kr-users.quwanzhi.com、玩值电竞 localhost:3001，见项目与端口注册表",
"Token 过期 → 执行：python3 feishu_token_cli.py get-access-token",
],
"t_thoughts": [
"Token 过期无需询问，直接命令刷新；上周总结优化便于周复盘闭环",
],
"w_work": [
"后台数据链接登记/验证",
"上周 3 月总结检查并优化进度",
"飞书日志写入",
],
"f_feedback": [
"后台数据链接 见 00_账号与API索引、项目与端口注册表",
"上周总结 已检查优化",
"Token 已刷新",
"今天结束前给出一个「200 条/日」的可执行方案（包含步骤、工具、人天）",
"进度评估：按「方案清晰度 + 工具可用度」两个维度持续更新百分比",
],
},
]