#!/usr/bin/env python3 """Extract weekly report material from a large xlsx without loading styles. The source workbook currently fails in openpyxl because of style metadata. This parser reads the xlsx XML directly and only extracts cell values. """ from __future__ import annotations import csv import hashlib import html import re import sys from dataclasses import dataclass from datetime import datetime, timezone, timedelta from pathlib import Path, PurePosixPath from typing import Dict, Iterable, List, Tuple from zipfile import ZipFile from xml.etree import ElementTree as ET MAIN_NS = "http://schemas.openxmlformats.org/spreadsheetml/2006/main" REL_NS = "http://schemas.openxmlformats.org/officeDocument/2006/relationships" PKG_REL_NS = "http://schemas.openxmlformats.org/package/2006/relationships" NS = { "m": MAIN_NS, "rel": REL_NS, "pkgrel": PKG_REL_NS, } CELL_REF_RE = re.compile(r"([A-Z]+)(\d+)") DATED_SHEET_RE = re.compile(r"^\d{8}-\d{8}$") @dataclass class Sheet: name: str sheet_id: str path: str def col_to_idx(col: str) -> int: value = 0 for ch in col: value = value * 26 + ord(ch) - 64 return value - 1 def normalize_text(value: str) -> str: return ( value.replace("\ufffc", "") .replace("\u00a0", " ") .replace("\r\n", "\n") .replace("\r", "\n") .strip() ) def sha256_file(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def load_shared_strings(zf: ZipFile) -> List[str]: if "xl/sharedStrings.xml" not in zf.namelist(): return [] strings: List[str] = [] with zf.open("xl/sharedStrings.xml") as fh: for event, elem in ET.iterparse(fh, events=("end",)): if elem.tag != f"{{{MAIN_NS}}}si": continue parts = [t.text or "" for t in elem.iter(f"{{{MAIN_NS}}}t")] strings.append(normalize_text("".join(parts))) elem.clear() return strings def load_sheets(zf: ZipFile) -> List[Sheet]: workbook = ET.fromstring(zf.read("xl/workbook.xml")) rels = ET.fromstring(zf.read("xl/_rels/workbook.xml.rels")) rid_to_target = { rel.attrib["Id"]: rel.attrib["Target"] for rel in rels.findall("pkgrel:Relationship", NS) } sheets: List[Sheet] = [] for sheet in workbook.findall(".//m:sheet", NS): rid = sheet.attrib[f"{{{REL_NS}}}id"] target = rid_to_target[rid] path = str(PurePosixPath("xl") / target) sheets.append(Sheet(sheet.attrib["name"], sheet.attrib["sheetId"], path)) return sheets def iter_sheet_rows( zf: ZipFile, path: str, shared_strings: List[str] ) -> Iterable[Tuple[int, Dict[int, str]]]: with zf.open(path) as fh: for event, row in ET.iterparse(fh, events=("end",)): if row.tag != f"{{{MAIN_NS}}}row": continue values: Dict[int, str] = {} for cell in row.findall(f"{{{MAIN_NS}}}c"): ref = cell.attrib.get("r", "") match = CELL_REF_RE.match(ref) if not match: continue idx = col_to_idx(match.group(1)) cell_type = cell.attrib.get("t") value = "" if cell_type == "inlineStr": value = "".join(t.text or "" for t in cell.iter(f"{{{MAIN_NS}}}t")) else: value_node = cell.find(f"{{{MAIN_NS}}}v") if value_node is not None and value_node.text is not None: raw = value_node.text if cell_type == "s": try: value = shared_strings[int(raw)] except (ValueError, IndexError): value = raw else: value = raw value = normalize_text(str(value)) if value: values[idx] = value if values: yield int(row.attrib.get("r", "0")), values row.clear() def write_csv(path: Path, rows: List[Dict[str, str]], fields: List[str]) -> None: path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=fields) writer.writeheader() writer.writerows(rows) def extract_summary_items( zf: ZipFile, sheet: Sheet, shared_strings: List[str] ) -> Tuple[str, List[Dict[str, str]], List[Dict[str, str]]]: rows = list(iter_sheet_rows(zf, sheet.path, shared_strings)) header = next(values for row_no, values in rows if row_no == 2) periods = {idx: value for idx, value in header.items() if idx >= 2} current_period = periods.get(2, "") current_items: List[Dict[str, str]] = [] all_items: List[Dict[str, str]] = [] current_person = "" for row_no, values in rows: if row_no <= 2: continue if values.get(0): current_person = values[0] category = values.get(1, "") if not current_person or not category: continue for idx, period in periods.items(): text = values.get(idx, "") if not text: continue row = { "sheet": sheet.name, "row": str(row_no), "person": current_person, "category": category, "period": period, "text": text, } all_items.append(row) if idx == 2: current_items.append(row) return current_period, current_items, all_items def extract_product_rnd( zf: ZipFile, sheet: Sheet, shared_strings: List[str] ) -> List[Dict[str, str]]: out: List[Dict[str, str]] = [] current_stage = "" for row_no, values in iter_sheet_rows(zf, sheet.path, shared_strings): if row_no < 5: continue if values.get(1): current_stage = values[1] if not values.get(2) and not values.get(3): continue out.append( { "sheet": sheet.name, "row": str(row_no), "stage": current_stage, "category": values.get(2, ""), "description": values.get(3, ""), "remark": values.get(8, ""), } ) return out def extract_dated_project_rows( zf: ZipFile, sheets: List[Sheet], shared_strings: List[str] ) -> List[Dict[str, str]]: out: List[Dict[str, str]] = [] for sheet in sheets: if not DATED_SHEET_RE.match(sheet.name): continue current_project_type = "" current_phase = "" for row_no, values in iter_sheet_rows(zf, sheet.path, shared_strings): if row_no <= 5: continue if values.get(1): current_project_type = values[1] if values.get(2): current_phase = values[2] project_name = values.get(3, "") if not project_name: continue out.append( { "period": sheet.name, "row": str(row_no), "project_type": current_project_type, "phase": current_phase, "project_name": project_name, "deadline_or_time": values.get(4, ""), "amount": values.get(5, ""), "owner": values.get(6, ""), "last_progress": values.get(7, ""), "this_week_and_next_plan": values.get(8, ""), "issue_or_deviation": values.get(9, ""), "remark": values.get(10, ""), "link": values.get(11, ""), } ) return out def count_nonempty_rows(zf: ZipFile, sheet: Sheet, shared_strings: List[str]) -> int: return sum(1 for _ in iter_sheet_rows(zf, sheet.path, shared_strings)) def report_markdown(current_period: str, source_path: Path) -> str: return f"""# 数字技术中心与交付实施部智慧食堂项目周报（2026-W20） > 周期：{current_period} > 来源：`{source_path}` > 证据等级：B。已解析团队 Excel 周报中的人员周输入；交付实施部未在 Excel 中作为独立字段拆出，本文将现场交付、上线联调、客户沟通、设备对接、培训和验收支持类事项归入“交付实施协同”。 ## 一、本周总体判断 本周智慧食堂相关工作围绕两条主线推进：一是保障当前项目交付和客户响应，二是把项目中的共性能力沉淀为标准产品、硬件能力、AI 工程化方法和售前交付资产。 - **当前项目需要**：金斯瑞、江西 206、首通智诚、江苏国信、苏州福耀、网信办、三全、大庆油田、赛迪等项目同步推进，重点集中在需求确认、上线验证、接口联调、设备对接、投标支持和客户问题闭环。 - **技术和产品引领**：教委食安监管大屏、教委总览大屏、进销存大屏、学校版产品材料、智慧食堂售前轻采集系统、AI 菜品识别、取餐柜/售卖柜链路、OA 补贴接口、设备兼容和 AI 协作流程开始形成可复用资产。 - **组织协同变化**：AI 已从资料整理和单点代码辅助，进一步进入需求拆解、原型生成、代码 review、接口文档、云效任务流转、测试数据和文档沉淀环节，但仍需要产品设计 SOP 和项目优先级机制来约束质量。 ## 二、满足当前项目需要 ### 1. 重点项目交付与客户响应 - **金斯瑞项目**：完成 2 个现场需求跟进、开发与上线；7 个新增问题已完成分析与回复；取餐柜、麻辣烫称打印小票等需求已提交云效并补充分支与异常流程；上线功能验收通过。同步推进 OA 对接接口开发、自测和上线，补贴余额查询、交易明细接口和接口文档继续完善。AI 菜品识别持续跟踪识别效果，5 月 14 日识别率记录达到 90% 以上，同时排查样本质量、出餐稳定性和相似菜误判问题。 - **江西 206 项目**：围绕售卖柜对接卡号问题进行多轮沟通，持续处理客户现场反馈；下周重点进入消费机离线消费和双秤开发，售卖柜问题继续按接口、设备、PC 订单和异常场景拆分推进。 - **首通智诚荣海小学项目**：完成核心需求确认，明确费用管控目标，完成缴费需求梳理和原型页面产出；小程序风险修复和代码加固方案已提供给客户；银行对接业务流程说明和流程图已沉淀，便于研发、业务和客户共同理解。 - **江苏国信项目**：完成合同内容和核心需求梳理，经过多轮线上会议明确整体需求与个性化需求并输出报价单；下周继续跟进系统部署、个性化需求和食安系统优化。 - **苏州福耀玻璃食堂项目**：完成项目细节沟通，确认食堂场景个性化需求，为后续售前方案、设备点位和交付拆解打基础。 - **网信办 / 三全 / 大庆 / 赛迪等项目**：网信办继续推进询价单发布、等保测试范围建议、供应商系统使用反馈和先进先出需求沟通；三全项目完成投标答疑和技术方案配合；大庆油田完成报价表功能描述优化；赛迪新增问题已完成需求分析并提交云效。 ### 2. 交付实施协同 - 交付实施侧本周重点不是单纯“到现场”，而是把客户现场需求转成可执行的产品、接口、设备和验证任务：金斯瑞取餐柜、江西 206 售卖柜、国信系统部署、教委大屏上线支撑、网信办进销存使用培训和等保沟通，均已进入研发与交付协同闭环。 - 针对食安和硬件交付，已开始关注设备兼容性、摄像头更换、烟感/燃气传感器、网关、留样柜、农残检测仪、消费机、双秤、厨房秤、取餐语音叫号等真实现场链路，后续需把设备协议、异常处理、测试记录和操作手册继续沉淀为交付资产。 ## 三、做好技术和产品引领 ### 1. 智慧食堂标准产品推进 - 学校版产品 PPT 已完成多轮更新，补充学生营养餐相关知识、业务流程和说明，后续可用于学校场景售前和内部统一口径。 - 近期全量项目需求已完成评审和拆解，多个项目新增需求已提交云效，需求从“口头响应”逐步进入可追踪的产品需求池。 - 教委食安监管大屏、教委总览大屏和进销存大屏持续优化：完成 3D 地图点位联动、数据卡片弹窗、模块悬浮等交互；将 HTML 原型样式改造成 Vue 开发样式，并把总览大屏与食安大屏整合到同一项目，降低后续维护成本。 - 标准版小程序完成首页跳转逻辑、付款码路径、登录态恢复和 AI 运动营养师播报优化，部分能力已完成自测和上线。 ### 2. 食安、进销存和监管场景延展 - 食安产品不再停留在后厨单点监管，正在向“环境监测 + 监管大屏 + 采购库存监管 + 指标算法 + 多场景数据支撑”的综合监管产品形态延展。 - 教委端产品开始围绕校园食安总览、监管平台总览和进销存大屏构建多角色、多层级监管视角，适合支撑学校、教委和政府监管类项目复制。 - 网信办项目继续作为进销存真实业务验证场景，通过供应商培训、报表沟通、询价单、比价和先进先出需求，把项目陪跑转为产品规则沉淀。 ### 3. AI 工程化与研发提效 - AI 已开始贯穿“需求抽象 -> 产品设计 -> 原型生成 -> 开发实现 -> review -> 测试 -> 文档沉淀”的链路。教委端大屏已采用 ChatGPT 生成角色、人设和设计 prompt，再由 Codex 辅助原型和样式优化。 - 云效需求处理正在尝试由 Codex 自动拉取需求、拆解任务、预估工时、关联代码和流转状态，减少人工维护成本。 - 金斯瑞取餐柜开始按“方案梳理 -> Codex 协作开发 -> 代码 review -> 测试”的方式推进，后续可复制到售卖柜、OA 对接、设备端需求。 - DeepSeek-TUI、Codex、Apifox 打通、多模态插件、健康数据 API、手环/厨房秤接入等方向已进入调研或验证，后续需要筛选出能落地到智慧食堂产品的能力。 ## 四、问题与风险 | 风险 | 影响 | 应对动作 | | --- | --- | --- | | AI 产品设计方法还不稳定 | 生成原型和需求说明可能看起来完整，但异常流程、现场分支和交付可用性不足 | 建立产品设计与评审 SOP，重要需求必须补主流程、异常流程、角色权限、数据边界和验收方式 | | 多项目并行造成资源切换 | 金斯瑞、江西 206、国信、教委大屏、网信办等并行，容易影响上线验证和问题闭环 | 按客户影响和上线节点排序，重点项目保留主负责人和每日状态同步 | | 现场设备链路依赖多 | 摄像头、传感器、网关、取餐柜、售卖柜、双秤、消费机等涉及供应商和现场环境 | 每个设备链路形成接口、配置、异常、测试和操作手册清单 | | 标准产品与项目定制边界需继续收敛 | 项目需求多，若全部定制会削弱标准化节奏 | 需求评审时明确“标准版吸收、项目版配置、临时定制、暂缓”四类结论 | ## 五、下周重点 1. **保障重点项目交付**：金斯瑞取餐柜上线验证、现场问题跟进；江西 206 离线消费和双秤开发；江苏国信个性化需求与系统部署；苏州福耀售前方案和需求拆解继续完善。 2. **推进监管和食安大屏产品化**：完成教委端食安、进销存、总览大屏页面优化和上线支撑，持续收敛展示内容和交互逻辑。 3. **补齐硬件交付资产**：围绕食安设备兼容、取餐语音叫号、摄像头、厨房秤、传感器、售卖柜和双秤，沉淀设备接入、测试和异常处理文档。 4. **建立产品设计与评审 SOP**：把学校版产品、智慧食堂业务 PPT、校园版大屏、现金退款、标准产品细节优化等内容纳入统一评审和版本管理。 5. **把 AI 提效固化为团队流程**：继续推进 Codex 与 Apifox、云效、代码 review、需求说明、接口文档和测试数据的打通，把有效做法沉淀为可复制的 Prompt、脚本和团队规范。 ## 六、待确认项 - Excel 中未单独提供“交付实施部”组织字段，本文根据项目交付、现场联调、上线支持、设备对接和客户沟通类事项归纳为交付实施协同。 - 部分项目缺少客户验收签字、上线截图、设备测试记录和最终交付状态，当前周报按“已推进 / 已完成本周动作 / 下周继续跟进”表达，不写成最终验收完成。 """ def report_html(markdown_text: str, current_period: str) -> str: # Simple purpose-built renderer for the known Markdown structure. body_lines: List[str] = [] in_table = False for line in markdown_text.splitlines(): if line.startswith("| 风险 |"): in_table = True body_lines.append("") continue if in_table and line.startswith("| ---"): continue if in_table and line.startswith("| "): cells = [html.escape(c.strip()) for c in line.strip("|").split("|")] body_lines.append("" + "".join(f"" for c in cells) + "") continue if in_table: body_lines.append("
风险影响应对动作
{c}
") in_table = False if line.startswith("# "): body_lines.append(f"

{html.escape(line[2:])}

") elif line.startswith("## "): body_lines.append(f"

{html.escape(line[3:])}

") elif line.startswith("### "): body_lines.append(f"

{html.escape(line[4:])}

") elif line.startswith("> "): body_lines.append(f"

{html.escape(line[2:])}

") elif line.startswith("- "): body_lines.append(f"

• {html.escape(line[2:])}

") elif re.match(r"^\d+\. ", line): body_lines.append(f"

{html.escape(line)}

") elif line.strip(): body_lines.append(f"

{html.escape(line)}

") if in_table: body_lines.append("") body = "\n".join(body_lines) return f""" 数字技术中心与交付实施部智慧食堂项目周报
2026-W20 · {html.escape(current_period)} · 智慧食堂项目周报

数字技术中心与交付实施部智慧食堂项目周报

围绕“满足当前项目需要”和“做好技术与产品引领”两条主线，整理团队 Excel 周报中的项目、产品、硬件、AI 和交付协同进展。

当前项目闭环金斯瑞、江西206、首通智诚、国信、苏州福耀等并行推进。
产品标准化学校版材料、教委大屏、食安与进销存能力持续沉淀。
AI 工程化AI 进入需求拆解、原型、开发、review、云效和文档沉淀。
{body}
""" def main() -> int: if len(sys.argv) != 3: print("usage: extract_xlsx_weekly_report.py ", file=sys.stderr) return 2 source = Path(sys.argv[1]).expanduser().resolve() output_dir = Path(sys.argv[2]).resolve() output_dir.mkdir(parents=True, exist_ok=True) generated_at = datetime.now(timezone(timedelta(hours=8))).strftime("%Y-%m-%d %H:%M %z") with ZipFile(source) as zf: shared_strings = load_shared_strings(zf) sheets = load_sheets(zf) sheet_index: List[Dict[str, str]] = [] for sheet in sheets: sheet_index.append( { "sheet_name": sheet.name, "sheet_id": sheet.sheet_id, "xml_path": sheet.path, "nonempty_row_count": str(count_nonempty_rows(zf, sheet, shared_strings)), } ) sheet_by_name = {sheet.name: sheet for sheet in sheets} current_period, current_items, all_summary_items = extract_summary_items( zf, sheet_by_name["2026年周报"], shared_strings ) product_rnd = extract_product_rnd(zf, sheet_by_name["产品研发（每周更新）"], shared_strings) dated_projects = extract_dated_project_rows(zf, sheets, shared_strings) source_index = [ { "source_type": "xlsx", "source_path": str(source), "sha256": sha256_file(source), "sheet_count": str(len(sheets)), "current_period": current_period, "generated_at": generated_at, "notes": "Parsed xlsx XML directly because openpyxl failed on workbook styles.", } ] write_csv( output_dir / "source-index.csv", source_index, ["source_type", "source_path", "sha256", "sheet_count", "current_period", "generated_at", "notes"], ) write_csv( output_dir / "workbook-sheet-index.csv", sheet_index, ["sheet_name", "sheet_id", "xml_path", "nonempty_row_count"], ) write_csv( output_dir / "team-current-week-items.csv", current_items, ["sheet", "row", "person", "category", "period", "text"], ) write_csv( output_dir / "all-2026-summary-items.csv", all_summary_items, ["sheet", "row", "person", "category", "period", "text"], ) write_csv( output_dir / "product-rd-summary.csv", product_rnd, ["sheet", "row", "stage", "category", "description", "remark"], ) write_csv( output_dir / "dated-project-rows.csv", dated_projects, [ "period", "row", "project_type", "phase", "project_name", "deadline_or_time", "amount", "owner", "last_progress", "this_week_and_next_plan", "issue_or_deviation", "remark", "link", ], ) md = report_markdown(current_period, source) (output_dir / "smart-canteen-weekly-report.md").write_text(md, encoding="utf-8") (output_dir / "smart-canteen-weekly-report.html").write_text( report_html(md, current_period), encoding="utf-8" ) people = sorted({row["person"] for row in current_items}) summary = f"""# 2026-W20 Excel 周报解析与智慧食堂专项周报生成记录 ## Goal 把 `/Users/jack/Downloads/数字技术中心工作周报2025-2026.xlsx` 中大家的周报内容解析后放入 `zhctprompt`，并再次生成数字技术中心与交付实施部在智慧食堂项目上的专项周报。 ## Source Context - 源文件：`{source}` - 文件 SHA256：`{source_index[0]["sha256"]}` - 工作簿 sheet 数：{len(sheets)} - 当前周周期：{current_period} - 解析方式：直接读取 xlsx XML，绕过 openpyxl 样式解析异常。 ## Parsed Outputs - `source-index.csv`：源文件、哈希、sheet 数和解析说明。 - `workbook-sheet-index.csv`：90 个 sheet 的名称、xml 路径和非空行统计。 - `team-current-week-items.csv`：当前周按人员、类别抽取的团队周输入，共 {len(current_items)} 条。 - `all-2026-summary-items.csv`：`2026年周报` 主表所有周期的人员周输入，共 {len(all_summary_items)} 条。 - `product-rd-summary.csv`：产品研发长期条目，共 {len(product_rnd)} 条。 - `dated-project-rows.csv`：历史日期 sheet 中按项目抽取的进度条目，共 {len(dated_projects)} 条。 - `smart-canteen-weekly-report.md`：AI 可读周报正文。 - `smart-canteen-weekly-report.html`：给用户 review 和转发的 HTML 周报。 ## Current Week People {", ".join(people)} ## Scope And Non-Goals - 本次不复制 211MB 原始 Excel 进仓库，只保存源路径、哈希和解析后的结构化内容。 - 不把缺少验收截图、客户签字或设备测试记录的事项写成最终验收完成。 - Excel 未单列“交付实施部”，报告中只把现场、联调、上线、设备和客户沟通类事项归为交付实施协同。 ## Human Review Status - 状态：待 Jack review。 - 重点 review：是否需要把报告口径从“数字技术中心与交付实施协同”改成正式组织名“数字技术中心和交付实施部”；是否补充交付实施部独立输入。 """ (output_dir / "summary.md").write_text(summary, encoding="utf-8") print(f"wrote {output_dir}") print(f"current_period={current_period}") print(f"current_items={len(current_items)} all_summary_items={len(all_summary_items)} dated_projects={len(dated_projects)}") return 0 if __name__ == "__main__": raise SystemExit(main())