Initial panel project deploy skill snapshot

This commit is contained in:
depei
2026-10-10 18:50:57 +08:00
commit 5295c1cf58
11 changed files with 2286 additions and 0 deletions
+169
View File
@@ -0,0 +1,169 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
交付压缩包构建脚本(合同对齐)。
产物(位于 --out-dir,默认 .Codex/delivery/):
1) <zip-name> 给甲方/终端使用的交付压缩包
2) 交付清单.md 与 zip 并列的单独文件(不打入包内)
zip 内部固定结构:
README.md 包根目录唯一一份 README
docs/screenshots/ README 引用的全部配图(解压可显示)
截图/界面截图/ 当前产品界面截图(中性文件名)
录屏/
文档/环境依赖清单.md
文档/部署注意事项.md
用法见 --help。
"""
from __future__ import annotations
import argparse
import os
import re
import shutil
import sys
import tempfile
import zipfile
from pathlib import Path
def copy_file(src: Path, dst: Path):
dst.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, dst)
def main() -> int:
ap = argparse.ArgumentParser(description="构建交付压缩包与并列交付清单")
ap.add_argument("--project-dir", required=True)
ap.add_argument("--readme", default="README.md")
ap.add_argument("--out-dir", default=".Codex/delivery")
ap.add_argument("--assets-product", default=".Codex/delivery/assets/product")
ap.add_argument("--recordings", default=".Codex/delivery/recordings")
ap.add_argument("--dep-doc", default=".Codex/delivery/环境依赖清单.md")
ap.add_argument("--notes-doc", default=".Codex/delivery/部署注意事项.md")
ap.add_argument("--manifest-src", default=".Codex/delivery/manifest.md")
ap.add_argument("--zip-name", default=None, help="默认 smart-parking-delivery-<YYYYMMDD>.zip")
ap.add_argument("--video-name", default="full-feature-demo.mp4")
args = ap.parse_args()
root = Path(args.project_dir).resolve()
out_dir = root / args.out_dir
out_dir.mkdir(parents=True, exist_ok=True)
readme_path = root / args.readme
if not readme_path.exists():
print("[FAIL] README 不存在:%s" % readme_path)
return 1
zip_name = args.zip_name or ("smart-parking-delivery-%s.zip" % _today())
zip_path = out_dir / zip_name
with tempfile.TemporaryDirectory() as td:
stage = Path(td)
# 1) README(包根唯一一份)
copy_file(readme_path, stage / "README.md")
# 2) README 引用图片 -> docs/...(保持相对路径)
readme_text = readme_path.read_text(encoding="utf-8", errors="ignore")
img_refs = re.findall(r"!\[[^\]]*\]\(([^)]+)\)", readme_text)
copied_img = set()
for ref in img_refs:
ref = ref.strip()
if "://" in ref or ref.startswith("#"):
continue
src_img = (readme_path.parent / ref).resolve()
if not src_img.exists():
print("[FAIL] README 引用图片缺失:%s" % ref)
return 1
rel = src_img.relative_to(root)
if str(rel) not in copied_img:
copy_file(src_img, stage / rel)
copied_img.add(str(rel))
# 3) 产品界面截图(中性命名,不体现版本对比或来源宣传)
product_assets = root / args.assets_product
if product_assets.is_dir():
shutil.copytree(product_assets, stage / "截图" / "界面截图", dirs_exist_ok=True)
# 4) 录屏(只取指定 mp4)
rec_dir = root / args.recordings
video = rec_dir / args.video_name
if video.exists():
copy_file(video, stage / "录屏" / args.video_name)
# 5) 文档目录:仅依赖清单 + 注意事项(无 README、无交付清单)
for d in (args.dep_doc, args.notes_doc):
p = root / d
if p.exists():
copy_file(p, stage / "文档" / p.name)
# 打包
if zip_path.exists():
zip_path.unlink()
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED, compresslevel=6) as z:
for p in sorted(stage.rglob("*")):
if p.is_file():
z.write(p, p.relative_to(stage).as_posix())
# 交付清单:复制为与 zip 并列的单独文件
manifest_src = root / args.manifest_src
checklist_path = out_dir / "交付清单.md"
if manifest_src.exists():
shutil.copy2(manifest_src, checklist_path)
# 校验:解压核对 README 图片
bad = _verify_zip(zip_path)
if bad:
print("[FAIL] 交付包校验未通过:%s" % bad)
return 1
print("交付包:%s (%.2f MB)" % (zip_path, zip_path.stat().st_size/1024/1024))
if checklist_path.exists():
print("并列交付清单:%s" % checklist_path)
return 0
def _today() -> str:
from datetime import datetime
return datetime.now().strftime("%Y%m%d")
def _verify_zip(zip_path: Path) -> list[str]:
problems: list[str] = []
with tempfile.TemporaryDirectory() as td:
with zipfile.ZipFile(zip_path) as z:
z.extractall(td)
v = Path(td)
names = [p.relative_to(v).as_posix() for p in v.rglob("*") if p.is_file()]
readmes = [n for n in names if n.lower().endswith("readme.md")]
if readmes != ["README.md"]:
problems.append("README 必须且只能在包根一份,实际:%s" % readmes)
if any("交付清单" in n for n in names):
problems.append("交付清单不得打入包内")
if any(re.search(r"(before|after|改造前|原版)", n, re.I) for n in names):
problems.append("包内不得出现对比期/before/after 文件命名")
docs = [n for n in names if n.startswith("文档/")]
if any(n.lower().endswith("readme.md") for n in docs):
problems.append("文档目录不得放 README")
if (v / "README.md").exists():
txt = (v / "README.md").read_text(encoding="utf-8", errors="ignore")
for ref in re.findall(r"!\[[^\]]*\]\(([^)]+)\)", txt):
if "://" in ref or ref.startswith("#"):
continue
if not (v / ref).exists():
problems.append("README 图片解压后缺失:%s" % ref)
forbidden_in_readme = [
"致谢", "原始开源项目", "原项目链接", "原作者", "基于开源",
"He Rui", "CrazyHer", "campus-surveillance",
]
for term in forbidden_in_readme:
if term.lower() in txt.lower():
problems.append("README 出现不应主动展示的来源表述:%s" % term)
return problems
if __name__ == "__main__":
sys.exit(main())
+227
View File
@@ -0,0 +1,227 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
交付物阻断式核查脚本(对齐软件改造部署外包交付规范)。
检查项:
A. 7 项交付物:源码 / 当前产品截图 / 全功能录屏 / 网盘归档 / README / Docker 部署 / 依赖与注意事项
B. README 12 项必备内容(按标题关键词启发式匹配)
C. 痕迹/来源清理:对用户提供的商业标识与来源关键词做全文检索(交付物中不得命中)
D. 交付清单 .Codex/delivery/manifest.md 存在且 7 项条目齐全
退出码:0 全部通过;1 存在阻断项。
用法见 --help。
"""
from __future__ import annotations
import argparse
import json
import os
import re
import sys
from datetime import datetime
from pathlib import Path
README_KEYWORDS = {
"一句话简介": ["简介", "项目介绍"],
"详细功能描述": ["功能", "功能特性", "功能特点"],
"行业痛点": ["痛点"],
"技术优势": ["技术优势", "核心优势"],
"应用场景": ["应用场景", "使用场景", "场景"],
"目标行业与人群": ["目标行业", "适用行业", "适配行业", "目标用户", "适用人群"],
"商业应用价值": ["商业价值", "应用价值"],
"技术架构(含拓扑图)": ["技术架构", "架构"],
"环境前置依赖": ["环境要求", "前置依赖", "依赖要求", "环境依赖"],
"快速开始": ["快速开始", "快速上手", "quick start"],
"安装部署教程": ["部署", "安装"],
"版权与使用说明": ["版权", "使用说明"],
}
SCAN_SUFFIXES = {
".ts", ".tsx", ".js", ".jsx", ".vue", ".java", ".py", ".go", ".json",
".md", ".html", ".yml", ".yaml", ".less", ".css", ".scss", ".xml",
}
IMAGE_SUFFIXES = {".png", ".jpg", ".jpeg", ".webp"}
VIDEO_SUFFIXES = {".mp4", ".mkv", ".mov", ".avi", ".webm"}
SKIP_DIRS = {
"node_modules", ".git", "dist", "build", "target", ".next", ".umi",
"__pycache__", ".venv", "venv",
}
def iter_files(root: Path):
# os.walk 自顶向下并在目录层剪枝,避免遍历 node_modules/.git 等巨型目录
for dirpath, dirnames, filenames in os.walk(root):
dirnames[:] = [
d for d in dirnames
if d not in SKIP_DIRS and not d.lower().startswith(".codex")
]
for name in filenames:
yield Path(dirpath) / name
def find_files(root: Path, suffixes: set[str]) -> list[Path]:
return [p for p in iter_files(root) if p.suffix.lower() in suffixes]
def main() -> int:
ap = argparse.ArgumentParser(description="开源改造项目交付物阻断式核查")
ap.add_argument("--project-dir", required=True, help="项目根目录")
ap.add_argument("--readme", default="README.md", help="README 相对路径")
ap.add_argument("--delivery-dir", default=".Codex/delivery", help="交付目录(相对项目)")
ap.add_argument("--assets-dir", default=".Codex/delivery/assets",
help="截图/录屏素材目录(相对项目)")
ap.add_argument("--trace-keywords", default="",
help="逗号分隔的商业标识与来源宣传关键词(原公司名、外部仓库、项目代号等)")
ap.add_argument("--allow-no-docker", action="store_true",
help="项目确实不适配 Docker 时,要求存在书面说明而非 Dockerfile")
ap.add_argument("--netdisk-link", default="",
help="网盘归档链接(manifest 中也可提供,二选一)")
ap.add_argument("--out", default=".Codex/reports/deliverables-check.json")
args = ap.parse_args()
root = Path(args.project_dir).resolve()
violations: list[str] = []
warnings: list[str] = []
evidence: dict = {}
# ---------- A1. 源码可构建产物线索 ----------
# 多模块项目(如 frontend/backend 分离)清单可能在子目录,递归查找
manifest_names = {
"package.json", "pom.xml", "build.gradle", "go.mod",
"requirements.txt", "pyproject.toml", "Cargo.toml",
}
manifest_hits = [p for p in iter_files(root) if p.name in manifest_names]
if not manifest_hits:
violations.append("未发现项目清单文件(package.json/pom.xml/go.mod 等),无法确认源码完整性")
else:
evidence["manifests"] = sorted(str(p.relative_to(root)) for p in manifest_hits)
# ---------- A2/A3. 截图与录屏 ----------
assets_dir = root / args.assets_dir
alt_dirs = [root / args.delivery_dir / "screenshots", root / args.delivery_dir / "recordings"]
images: list[Path] = []
videos: list[Path] = []
search_dirs = [d for d in [assets_dir, *alt_dirs] if d.exists()]
for d in search_dirs:
images.extend(find_files(d, IMAGE_SUFFIXES))
videos.extend(find_files(d, VIDEO_SUFFIXES))
# 对外只提供当前产品界面截图;交付素材目录不得含改造前/before 旧界面
leaked_before = [p for p in images if re.search(r"(before|改造前|原版)", p.name, re.I)]
if leaked_before:
violations.append("交付素材中包含改造前/原版截图(不应随包交付):%s"
% "、".join(p.name for p in leaked_before[:8]))
if not images:
violations.append("缺少产品界面截图(应放在 %s 或交付目录)" % args.assets_dir)
else:
evidence["screenshots"] = len(images)
if not videos:
violations.append("缺少全功能操作演示录屏(mp4/mov 等)")
else:
evidence["recordings"] = len(videos)
# ---------- A5. README ----------
readme_path = root / args.readme
readme_text = ""
if not readme_path.exists():
violations.append("缺少 README 文档:%s" % args.readme)
else:
readme_text = readme_path.read_text(encoding="utf-8", errors="ignore")
missing = []
low = readme_text.lower()
for item, kws in README_KEYWORDS.items():
if not any(k.lower() in low for k in kws):
missing.append(item)
if missing:
violations.append("README 缺少必备内容:%s" % "、".join(missing))
# 架构图线索:mermaid 或图片引用
if not re.search(r"```mermaid|!\[.*\]\(.*\)|架构图|拓扑图", readme_text):
warnings.append("README 未发现架构/拓扑图(mermaid 或图片)")
# 截图图文对应
if not re.search(r"!\[.*\]\(.*\)", readme_text):
warnings.append("README 无图片引用,合同要求功能逐条对应截图")
# ---------- A6. Docker / 书面说明 ----------
docker_files = [p for p in [root / "Dockerfile", root / "docker-compose.yml", root / "docker-compose.yaml"] if p.exists()]
if not docker_files:
if args.allow_no_docker:
note = root / args.delivery_dir / "no-docker说明.md"
if not note.exists():
violations.append("已声明不适配 Docker,但缺少书面说明(%s)" % note.relative_to(root))
else:
violations.append("缺少 Dockerfile / docker-compose(不适配 Docker 请加 --allow-no-docker 并提供书面说明)")
# ---------- A7. 依赖清单与部署注意事项 ----------
dep_docs = list((root / args.delivery_dir).glob("*依赖*")) + list((root / args.delivery_dir).glob("*注意事项*"))
dep_docs += [root / "依赖清单.md", root / "部署注意事项.md"]
dep_docs = [p for p in dep_docs if p.exists()]
if not dep_docs:
# 若 README 已含环境依赖/注意事项章节,降级为警告
if re.search(r"环境(要求|依赖|前置)|注意事项", readme_text):
warnings.append("未单独提供依赖清单/部署注意事项文档(README 中有相关章节,建议单独输出)")
else:
violations.append("缺少环境依赖清单与部署注意事项说明文档")
# ---------- A4. 网盘链接 + D. manifest ----------
manifest_path = root / args.delivery_dir / "manifest.md"
manifest_text = manifest_path.read_text(encoding="utf-8", errors="ignore") if manifest_path.exists() else ""
netdisk = args.netdisk_link or (re.search(r"https?://\S*(?:pan\.baidu|aliyundrive|alipan|123pan|cloud\.189)\S*", manifest_text) or [""])[0]
if not netdisk:
if not manifest_text:
violations.append("缺少网盘归档链接,且未提供交付清单 manifest.md")
else:
violations.append("未提供网盘素材归档链接(命令行 --netdisk-link 或写入 manifest)")
if not manifest_path.exists():
violations.append("缺少交付清单:%s" % manifest_path.relative_to(root))
else:
required_items = ["源码", "截图", "录屏", "网盘", "README", "Docker", "依赖"]
miss_items = [w for w in required_items if w not in manifest_text]
if miss_items:
violations.append("manifest.md 未覆盖交付物条目:%s" % "、".join(miss_items))
# ---------- C. 痕迹清理检索 ----------
kws = [k.strip() for k in args.trace_keywords.split(",") if k.strip()]
traces: list[str] = []
if kws:
pat = re.compile("|".join(re.escape(k) for k in kws), re.I)
for p in iter_files(root):
if p.suffix.lower() not in SCAN_SUFFIXES:
continue
rel = str(p.relative_to(root))
if re.search(r"license|copying", rel, re.I):
continue
try:
txt = p.read_text(encoding="utf-8", errors="ignore")
except OSError:
continue
if pat.search(txt):
traces.append(rel)
if traces:
violations.append("痕迹清理不彻底,以下文件命中原标识关键词:%s" % "、".join(sorted(set(traces))[:20]))
# ---------- 输出 ----------
result = {
"timestamp": datetime.now().isoformat(),
"project_dir": str(root),
"violations": violations,
"warnings": warnings,
"evidence": evidence,
"passed": not violations,
}
out_path = root / args.out
out_path.parent.mkdir(parents=True, exist_ok=True)
out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
print("交付物核查:%d 个阻断项,%d 个警告" % (len(violations), len(warnings)))
for v in violations:
print(" [FAIL]", v)
for w in warnings:
print(" [WARN]", w)
print("report=%s" % out_path)
return 0 if not violations else 1
if __name__ == "__main__":
sys.exit(main())
+351
View File
@@ -0,0 +1,351 @@
#!/usr/bin/env python3
"""Blocking audit for dark glass-panel UIs.
Requires Playwright: python -m pip install playwright && playwright install chromium.
"""
from __future__ import annotations
import argparse
import json
import os
import sys
from pathlib import Path
from typing import Any
from playwright.sync_api import Page, sync_playwright
AUDIT_JS = r"""
(() => {
function parseColor(value) {
const match = /rgba?\(([^)]+)\)/.exec(value || '');
if (!match) return {r: 255, g: 255, b: 255, a: 1};
const parts = match[1].split(',').map((x) => Number(x.trim()));
return {r: parts[0], g: parts[1], b: parts[2], a: parts.length > 3 ? parts[3] : 1};
}
function composite(top, bottom) {
const alpha = top.a + (1 - top.a) * bottom.a;
if (alpha === 0) return {r: bottom.r, g: bottom.g, b: bottom.b, a: 0};
const channel = (t, b) => (top.a * t + (1 - top.a) * b) / alpha;
return {r: channel(top.r, bottom.r), g: channel(top.g, bottom.g), b: channel(top.b, bottom.b), a: alpha};
}
function splitTopLevel(text) {
const parts = [];
let depth = 0;
let start = 0;
for (let i = 0; i < text.length; i += 1) {
if (text[i] === '(') depth += 1;
if (text[i] === ')') depth -= 1;
if (text[i] === ',' && depth === 0) {
parts.push(text.slice(start, i).trim());
start = i + 1;
}
}
const last = text.slice(start).trim();
if (last) parts.push(last);
return parts;
}
function parseColorOrNull(value) {
const match = /rgba?\(([^)]*)\)/.exec(value || '');
if (!match) return null;
const parts = match[1].split(',').map((x) => Number(x.trim()));
return {r: parts[0], g: parts[1], b: parts[2], a: parts.length > 3 ? parts[3] : 1};
}
function averageColors(colors) {
if (!colors.length) return null;
const total = colors.reduce((acc, color) => ({
r: acc.r + color.r,
g: acc.g + color.g,
b: acc.b + color.b,
a: acc.a + color.a,
}), {r: 0, g: 0, b: 0, a: 0});
return {
r: total.r / colors.length,
g: total.g / colors.length,
b: total.b / colors.length,
a: Math.min(1, total.a / colors.length),
};
}
function extractGradient(text, keyword) {
const start = text.search(new RegExp(`(?:^|[^a-z-])${keyword}\\(`, 'i'));
if (start < 0) return null;
const open = text.indexOf('(', start + keyword.length - 1);
let depth = 0;
for (let i = open; i < text.length; i += 1) {
if (text[i] === '(') depth += 1;
if (text[i] === ')') {
depth -= 1;
if (depth === 0) return text.slice(open + 1, i);
}
}
return null;
}
function gradientAverage(backgroundImage, keyword) {
const content = extractGradient(backgroundImage, keyword);
if (!content) return null;
const colors = splitTopLevel(content)
.map((part) => parseColorOrNull(part))
.filter(Boolean);
return averageColors(colors);
}
function backgroundImageLayers(style) {
const image = style.backgroundImage;
if (!image || image === 'none') return [];
return ['conic-gradient', 'radial-gradient', 'linear-gradient']
.map((keyword) => gradientAverage(image, keyword))
.filter(Boolean);
}
function effectiveBackground(startEl) {
const chain = [];
let el = startEl;
while (el && el.nodeType === 1) {
const style = getComputedStyle(el);
chain.unshift({
color: parseColor(style.backgroundColor),
images: backgroundImageLayers(style),
});
el = el.parentElement;
}
let bg = {r: 255, g: 255, b: 255, a: 1};
for (const layer of chain) {
bg = composite(layer.color, bg);
for (const imageLayer of layer.images) bg = composite(imageLayer, bg);
}
return bg;
}
function channelLuminance(value) {
const c = Math.max(0, Math.min(255, value)) / 255;
return c <= 0.03928 ? c / 12.92 : Math.pow((c + 0.055) / 1.055, 2.4);
}
function luminance(color) {
return 0.2126 * channelLuminance(color.r) + 0.7152 * channelLuminance(color.g) + 0.0722 * channelLuminance(color.b);
}
function contrastRatio(first, second) {
const a = luminance(first);
const b = luminance(second);
return (Math.max(a, b) + 0.05) / (Math.min(a, b) + 0.05);
}
function visible(el) {
const rect = el.getBoundingClientRect();
const style = getComputedStyle(el);
return rect.width >= 2 && rect.height >= 2 &&
style.display !== 'none' && style.visibility !== 'hidden' &&
Number(style.opacity) >= 0.1;
}
function hasDirectText(el) {
return Array.from(el.childNodes).some((node) => node.nodeType === 3 && node.textContent.trim().length > 0);
}
function isMedia(el) {
return Boolean(el.closest && el.closest(
'img,picture,video,canvas,svg,iframe,.leaflet-container,[class*="leaflet"],[class*="map-container"],[class*="map-wrapper"],[data-map],[class*="player"]'
));
}
function inFloatingLayer(el) {
return Boolean(el.closest && el.closest(
'.ant-select-dropdown,.ant-picker-dropdown,.ant-modal-content,.ant-drawer-content,.ant-popover-inner,.ant-tooltip,.ant-cascader-dropdown,[role="dialog"],[role="alertdialog"]'
));
}
const bodyBg = effectiveBackground(document.body);
const bodyLuminance = luminance(bodyBg);
const issues = [];
const seen = new Set();
const nodes = Array.from(document.querySelectorAll('body *')).filter(visible).slice(0, 1800);
for (const el of nodes) {
const style = getComputedStyle(el);
const canCarryText = hasDirectText(el) || el.matches(
'input,textarea,button,select,a,label,summary,option,[role="button"],[role="option"],[role="menuitem"],[role="tab"],[aria-label]'
);
if (!canCarryText) continue;
const text = (el.innerText || el.value || el.placeholder || el.getAttribute('aria-label') || '').trim();
if (!text) continue;
const bg = effectiveBackground(el);
const styledColor = parseColor(style.color);
const fg = composite(styledColor, bg);
const ratio = contrastRatio(fg, bg);
const fontSize = Number.parseFloat(style.fontSize);
const bold = Number.parseInt(style.fontWeight, 10) >= 600;
const large = fontSize >= 18.5 || (bold && fontSize >= 15);
const required = large ? 3 : 4.5;
const rect = el.getBoundingClientRect();
function push(reason, severity) {
const item = {
reason,
severity,
text: text.slice(0, 140),
tag: el.tagName.toLowerCase(),
className: String(el.className || '').slice(0, 180),
id: el.id || '',
background: `rgb(${Math.round(bg.r)}, ${Math.round(bg.g)}, ${Math.round(bg.b)})`,
color: `rgb(${Math.round(fg.r)}, ${Math.round(fg.g)}, ${Math.round(fg.b)})`,
contrast: Number(ratio.toFixed(2)),
requiredContrast: required,
backgroundLuminance: Number(luminance(bg).toFixed(3)),
bodyLuminance: Number(bodyLuminance.toFixed(3)),
size: `${Math.round(rect.width)}x${Math.round(rect.height)}`,
floating: inFloatingLayer(el),
};
const key = JSON.stringify([item.reason, item.text, item.className, item.contrast, item.background]);
if (!seen.has(key)) {
seen.add(key);
issues.push(item);
}
}
if (ratio < required) push('text contrast below threshold', 'P0');
const bgLum = luminance(bg);
if (!isMedia(el) && bgLum >= 0.62 && bodyLuminance <= 0.16 && rect.width >= 32 && rect.height >= 20) {
push('light component embedded in dark shell', 'P0');
}
}
return {bodyBg, bodyLuminance, issues};
})
"""
FLOAT_OPENERS = [
'.ant-select:not(.ant-select-disabled) .ant-select-selector',
'.ant-picker input',
]
def route_url(base_url: str, route: str, hash_routes: bool) -> str:
base = base_url.rstrip('/')
route = route if route.startswith('/') else f'/{route}'
return f'{base}/#{route}' if hash_routes else f'{base}{route}'
def audit_page(page: Page, route: str, extra_openers: list[str]) -> dict[str, Any]:
issues: list[dict[str, Any]] = []
data = page.evaluate(AUDIT_JS)
issues.extend(item for item in data['issues'] if not item.get('floating'))
selectors = FLOAT_OPENERS + extra_openers
for selector in selectors:
try:
locator = page.locator(selector).first
if locator.count() == 0 or not locator.is_visible():
continue
locator.click(timeout=1200)
page.wait_for_timeout(500)
opened = page.evaluate(AUDIT_JS)
issues.extend(item for item in opened['issues'] if item.get('floating'))
except Exception:
pass
finally:
for _ in range(2):
page.keyboard.press('Escape')
page.wait_for_timeout(120)
return {'route': route, 'issues': issues}
def launch_chromium(playwright):
launch_options = {'headless': True}
channel = os.environ.get('PANEL_PLAYWRIGHT_CHANNEL')
executable_path = os.environ.get('PANEL_PLAYWRIGHT_EXECUTABLE_PATH')
if channel:
launch_options['channel'] = channel
elif executable_path:
launch_options['executable_path'] = executable_path
else:
local_app_data = os.environ.get('LOCALAPPDATA', '')
playwright_cache = Path(local_app_data) / 'ms-playwright' if local_app_data else None
has_downloaded_chromium = bool(
playwright_cache and list(playwright_cache.glob('chromium-*/chrome-win64/chrome.exe'))
)
if not has_downloaded_chromium:
program_files_x86 = os.environ.get('ProgramFiles(x86)', '')
program_files = os.environ.get('ProgramFiles', '')
edge_candidates = [
Path(program_files_x86) / 'Microsoft' / 'Edge' / 'Application' / 'msedge.exe',
Path(program_files) / 'Microsoft' / 'Edge' / 'Application' / 'msedge.exe',
]
if any(edge.exists() for edge in edge_candidates):
launch_options['channel'] = 'msedge'
return playwright.chromium.launch(**launch_options)
def csv_list(value: str) -> list[str]:
return [item.strip() for item in value.split(',') if item.strip()]
def main() -> int:
parser = argparse.ArgumentParser(description='Audit panel UI contrast and light-in-dark nesting.')
parser.add_argument('--base-url', required=True)
parser.add_argument('--routes', required=True, help='Comma-separated routes, for example /campusState,/alarms')
parser.add_argument('--hash-routes', action='store_true')
parser.add_argument('--login-route', default='/login')
parser.add_argument('--skip-login', action='store_true')
parser.add_argument('--username', default='')
parser.add_argument('--password', default='')
parser.add_argument('--username-selector', default='input[type="text"],input:not([type]),input[name="username"]')
parser.add_argument('--password-selector', default='input[type="password"]')
parser.add_argument('--submit-text', default='')
parser.add_argument('--submit-selector', default='button[type="submit"]')
parser.add_argument('--open-selector', action='append', default=[], help='Optional element that opens a drawer/modal; repeatable')
parser.add_argument('--viewport-width', type=int, default=1440)
parser.add_argument('--viewport-height', type=int, default=900)
parser.add_argument('--wait-until', default='domcontentloaded', choices=['load', 'domcontentloaded', 'networkidle', 'commit'])
parser.add_argument('--out', default='.Codex/reports/panel-ui-audit.json')
args = parser.parse_args()
routes = csv_list(args.routes)
report: dict[str, Any] = {'baseUrl': args.base_url, 'pages': []}
with sync_playwright() as playwright:
browser = launch_chromium(playwright)
page = browser.new_page(viewport={'width': args.viewport_width, 'height': args.viewport_height})
if not args.skip_login:
page.goto(route_url(args.base_url, args.login_route, args.hash_routes), wait_until=args.wait_until)
if args.username:
page.locator(args.username_selector).first.fill(args.username)
if args.password:
page.locator(args.password_selector).first.fill(args.password)
if args.submit_text:
page.get_by_role('button', name=args.submit_text).click()
else:
page.locator(args.submit_selector).first.click()
page.wait_for_load_state('networkidle')
page.wait_for_timeout(800)
for route in routes:
page.goto(route_url(args.base_url, route, args.hash_routes), wait_until=args.wait_until)
page.wait_for_timeout(1200)
report['pages'].append(audit_page(page, route, args.open_selector))
browser.close()
out = Path(args.out)
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding='utf-8')
total = sum(len(page_report['issues']) for page_report in report['pages'])
print(f'audited {len(report["pages"])} routes; {total} violations')
print(f'report={out.resolve()}')
return 1 if total else 0
if __name__ == '__main__':
sys.exit(main())
+570
View File
@@ -0,0 +1,570 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
项目初始化与部署工作流管理器
提供状态管理、步骤执行和错误恢复功能
"""
import os
import sys
import json
import shutil
import subprocess
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, asdict
from enum import Enum
class StepStatus(Enum):
PENDING = "⏭️ 未开始"
IN_PROGRESS = "⏸️ 进行中"
COMPLETED = "✅ 已完成"
FAILED = "❌ 失败"
SKIPPED = "⏭️ 跳过"
class ProjectType(Enum):
DOCKER_WEB = "docker_web"
DOCKER_API = "docker_api"
DOCKER_MICROSERVICES = "docker_microservices"
DESKTOP = "desktop"
MOBILE = "mobile"
CLI = "cli"
UNKNOWN = "unknown"
class TechStack(Enum):
JAVA_SPRING = "java_spring"
NODE_EXPRESS = "node_express"
NODE_NEST = "node_nest"
PYTHON_DJANGO = "python_django"
PYTHON_FASTAPI = "python_fastapi"
GO_GIN = "go_gin"
GO_FIBER = "go_fiber"
REACT = "react"
VUE = "vue"
ANGULAR = "angular"
UNKNOWN = "unknown"
@dataclass
class ProjectInfo:
"""项目信息"""
name: str
directory: str
git_url: str
port_range: Optional[str] = None
project_type: ProjectType = ProjectType.UNKNOWN
tech_stack: List[TechStack] = None
def __post_init__(self):
if self.tech_stack is None:
self.tech_stack = []
@dataclass
class Step:
"""工作流步骤"""
id: int
name: str
description: str
applicable_types: List[ProjectType]
status: StepStatus = StepStatus.PENDING
start_time: Optional[str] = None
end_time: Optional[str] = None
notes: str = ""
error_info: Optional[str] = None
class WorkflowManager:
"""工作流管理器"""
# 完整的20个核心步骤
ALL_TYPES = [ProjectType.DOCKER_WEB, ProjectType.DOCKER_API, ProjectType.DOCKER_MICROSERVICES, ProjectType.DESKTOP, ProjectType.MOBILE, ProjectType.CLI]
DOCKER_TYPES = [ProjectType.DOCKER_WEB, ProjectType.DOCKER_API, ProjectType.DOCKER_MICROSERVICES]
# 关键时序:初始基线启动(3,内部) → 初始界面截图(4,内部) → 面板改造(6-11) → 产品构建启动(15)
# → 健康验证(16) → 产品截图与录屏(17) → README/归档(18) → 交付核查(19)
CORE_STEPS = [
Step(0, "克隆项目", "从Git仓库克隆项目到本地,不修改任何业务代码", ALL_TYPES),
Step(1, "项目完整度评估", "评估项目完整度、依赖情况与可运行性", ALL_TYPES),
Step(2, "项目整体分析", "分析项目结构、技术栈、启动方式和部署策略", ALL_TYPES),
Step(3, "初始版本基线启动(内部)", "在改动任何代码前,用项目现有启动方式跑通初始版本并确认可访问;仅用于内部质量核对", ALL_TYPES),
Step(4, "初始界面基线截图(内部)", "代码零改动状态下采集初始界面作为内部对比基线;严禁进入交付包或使用 before 等对外文件名", ALL_TYPES),
Step(5, "Docker配置生成", "生成/完善 Docker Compose 配置文件", DOCKER_TYPES),
Step(6, "页面信息与品牌名称修改", "替换页面品牌信息并建立科技感面板基础样式", ALL_TYPES),
Step(7, "Logo生成与替换", "生成并替换项目 Logo、Favicon,适配深色面板", ALL_TYPES),
Step(8, "版权与第三方标识规范", "清理依法可移除的公司名称、LOGO、商标、水印、联系方式等专属标识;MIT 等许可证要求保留的版权与许可文本必须随源码保留", ALL_TYPES),
Step(9, "页脚、导航与社区链接", "导航、页脚改为半透明悬浮面板风格,并放置开源社区与 GCC AI Coding 链接", ALL_TYPES),
Step(10, "登录页面", "改成居中玻璃面板、发光输入框、动态背景", ALL_TYPES),
Step(11, "AI对话功能", "实现前端AI对话界面与指定服务商API对接及使用教程", ALL_TYPES),
Step(12, "示例数据", "生成和导入示例数据,数据卡片/列表面板承载", DOCKER_TYPES),
Step(13, "CORS配置检查", "检查并配置 CORS 设置", DOCKER_TYPES),
Step(14, "端口检查与分配", "检查端口占用并分配端口", DOCKER_TYPES),
Step(15, "改造后构建与启动", "Docker项目构建镜像启动容器;非Docker项目手动构建并启动改造版", ALL_TYPES),
Step(16, "健康验证", "验证服务健康、面板UI审计0违规、全功能走查无报错", ALL_TYPES),
Step(17, "产品截图与全功能录屏", "采集当前产品全套高清界面截图(中性文件名)和全功能演示录屏;内部基线素材不进入交付物", ALL_TYPES),
Step(18, "标准化README与版本归档", "编写标准化 README(引用最终素材、含架构图与部署教程);不主动写来源/致谢宣传,并按授权决定是否提交", ALL_TYPES),
Step(19, "交付物核查", "按7项交付物清单与7条验收标准做交付前阻断式核查", ALL_TYPES),
]
# 旧版本步骤名 → 新版本步骤名(用于历史状态迁移)
STEP_NAME_ALIASES = {
"构建与启动": "改造后构建与启动",
"素材录制归档": "产品截图与全功能录屏",
"原始项目基线启动": "初始版本基线启动(内部)",
"改造前基线截图": "初始界面基线截图(内部)",
"原作者信息与版权痕迹合规处理": "版权与第三方标识规范",
"页脚与导航增强": "页脚、导航与社区链接",
"改造后截图与全功能录屏": "产品截图与全功能录屏",
"标准化README与提交": "标准化README与版本归档",
}
def __init__(self, project_dir: str):
self.project_dir = Path(project_dir)
self.claude_dir = self.project_dir / ".Codex"
self.state_file = self.claude_dir / "workflow_state.json"
self.error_file = self.claude_dir / "error_log.md"
self.project_info: Optional[ProjectInfo] = None
self.steps: List[Step] = []
def init_project(self, name: str, git_url: str, port_range: Optional[str] = None):
"""初始化项目信息"""
self.project_info = ProjectInfo(
name=name,
directory=str(self.project_dir),
git_url=git_url,
port_range=port_range
)
self.steps = [Step(s.id, s.name, s.description, s.applicable_types) for s in self.CORE_STEPS]
self._ensure_directories()
self._save_state()
def _ensure_directories(self):
"""确保必要的目录存在"""
self.claude_dir.mkdir(exist_ok=True)
(self.claude_dir / 'reports').mkdir(exist_ok=True)
def _save_state(self):
"""保存状态到JSON文件"""
def enum_to_value(obj):
if isinstance(obj, Enum):
return obj.value
if isinstance(obj, list):
return [enum_to_value(item) for item in obj]
if isinstance(obj, dict):
return {k: enum_to_value(v) for k, v in obj.items()}
return obj
project_info_dict = None
if self.project_info:
project_info_dict = asdict(self.project_info)
project_info_dict["project_type"] = project_info_dict["project_type"].value
project_info_dict["tech_stack"] = [ts.value for ts in project_info_dict["tech_stack"]]
state = {
"project_info": project_info_dict,
"steps": [
{
"id": s.id,
"name": s.name,
"status": s.status.value,
"start_time": s.start_time,
"end_time": s.end_time,
"notes": s.notes,
"error_info": s.error_info
}
for s in self.steps
],
"last_updated": datetime.now().isoformat()
}
with open(self.state_file, 'w', encoding='utf-8') as f:
json.dump(state, f, ensure_ascii=False, indent=2)
def load_state(self) -> bool:
"""从JSON文件加载状态"""
if not self.state_file.exists():
return False
try:
with open(self.state_file, 'r', encoding='utf-8') as f:
state = json.load(f)
if state.get("project_info"):
info = state["project_info"]
self.project_info = ProjectInfo(
name=info["name"],
directory=info["directory"],
git_url=info["git_url"],
port_range=info.get("port_range"),
project_type=ProjectType(info.get("project_type", "unknown")),
tech_stack=[TechStack(ts) for ts in info.get("tech_stack", [])]
)
# 以新 CORE_STEPS 为权威骨架;旧状态按“步骤名(含别名)”映射进度,
# 避免编号错位导致已完成的步骤被重置。
old_steps = state.get("steps", [])
name_to_state: Dict[str, dict] = {}
for s in old_steps:
canon = self.STEP_NAME_ALIASES.get(s.get("name", ""), s.get("name", ""))
name_to_state[canon] = s
self.steps = []
for cs in self.CORE_STEPS:
prev = name_to_state.get(cs.name)
if prev is not None:
self.steps.append(Step(
id=cs.id,
name=cs.name,
description=cs.description,
applicable_types=cs.applicable_types,
status=StepStatus(prev.get("status", StepStatus.PENDING.value)),
start_time=prev.get("start_time"),
end_time=prev.get("end_time"),
notes=prev.get("notes", ""),
error_info=prev.get("error_info"),
))
else:
# 新增步骤默认未开始
self.steps.append(Step(cs.id, cs.name, cs.description, cs.applicable_types))
return True
except Exception as e:
self._log_error(f"加载状态失败: {e}")
return False
def get_current_step(self) -> Optional[Step]:
"""获取当前步骤"""
for step in self.steps:
if step.status == StepStatus.IN_PROGRESS:
return step
if step.status == StepStatus.PENDING:
return step
return None
def start_step(self, step_id: int) -> bool:
"""开始执行步骤"""
step = next((s for s in self.steps if s.id == step_id), None)
if not step:
return False
# 检查步骤是否适用于当前项目类型
if self.project_info and self.project_info.project_type != ProjectType.UNKNOWN:
if self.project_info.project_type not in step.applicable_types:
step.status = StepStatus.SKIPPED
step.notes = f"步骤不适用于项目类型: {self.project_info.project_type.value}"
self._save_state()
return True
step.status = StepStatus.IN_PROGRESS
step.start_time = datetime.now().isoformat()
self._save_state()
return True
def complete_step(self, step_id: int, notes: str = ""):
"""完成步骤"""
step = next((s for s in self.steps if s.id == step_id), None)
if step:
step.status = StepStatus.COMPLETED
step.end_time = datetime.now().isoformat()
step.notes = notes
self._save_state()
def fail_step(self, step_id: int, error_info: str):
"""标记步骤失败"""
step = next((s for s in self.steps if s.id == step_id), None)
if step:
step.status = StepStatus.FAILED
step.end_time = datetime.now().isoformat()
step.error_info = error_info
self._save_state()
self._log_error(f"步骤 {step_id} ({step.name}) 失败: {error_info}")
def skip_step(self, step_id: int, reason: str = ""):
"""跳过步骤"""
step = next((s for s in self.steps if s.id == step_id), None)
if step:
step.status = StepStatus.SKIPPED
step.notes = reason
self._save_state()
def _log_error(self, message: str):
"""记录错误到错误日志"""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
error_entry = f"\n## [{timestamp}]\n{message}\n\n"
with open(self.error_file, 'a', encoding='utf-8') as f:
f.write(error_entry)
def get_progress(self) -> Dict:
"""获取进度信息"""
total = len(self.steps)
completed = sum(1 for s in self.steps if s.status == StepStatus.COMPLETED)
failed = sum(1 for s in self.steps if s.status == StepStatus.FAILED)
skipped = sum(1 for s in self.steps if s.status == StepStatus.SKIPPED)
pending = sum(1 for s in self.steps if s.status == StepStatus.PENDING)
return {
"total": total,
"completed": completed,
"failed": failed,
"skipped": skipped,
"pending": pending,
"percentage": (completed / total * 100) if total > 0 else 0
}
def can_resume(self) -> Tuple[bool, Optional[int]]:
"""检查是否可以恢复执行"""
if not self.state_file.exists():
return False, None
self.load_state()
# 检查是否有失败的步骤
failed_step = next((s for s in self.steps if s.status == StepStatus.FAILED), None)
if failed_step:
return True, failed_step.id
# 检查是否有进行中的步骤
in_progress = next((s for s in self.steps if s.status == StepStatus.IN_PROGRESS), None)
if in_progress:
return True, in_progress.id
# 检查是否有未开始的步骤
pending = next((s for s in self.steps if s.status == StepStatus.PENDING), None)
if pending:
return True, pending.id
return False, None
def generate_report(self) -> str:
"""生成执行报告"""
progress = self.get_progress()
report = f"""# 工作流执行报告
## 项目信息
- **项目名称**: {self.project_info.name if self.project_info else 'N/A'}
- **项目目录**: {self.project_info.directory if self.project_info else 'N/A'}
- **Git地址**: {self.project_info.git_url if self.project_info else 'N/A'}
- **项目类型**: {self.project_info.project_type.value if self.project_info else 'N/A'}
## 执行进度
- **总步骤**: {progress['total']}
- **已完成**: {progress['completed']}
- **失败**: {progress['failed']}
- **跳过**: {progress['skipped']}
- **待执行**: {progress['pending']}
- **完成度**: {progress['percentage']:.1f}%
## 步骤详情
"""
for step in self.steps:
report += f"\n### 步骤 {step.id}: {step.name}\n"
report += f"- **状态**: {step.status.value}\n"
if step.start_time:
report += f"- **开始时间**: {step.start_time}\n"
if step.end_time:
report += f"- **完成时间**: {step.end_time}\n"
if step.notes:
report += f"- **说明**: {step.notes}\n"
if step.error_info:
report += f"- **错误**: {step.error_info}\n"
return report
# 技术栈检测工具
class TechStackDetector:
"""技术栈检测器"""
DETECTION_RULES = {
TechStack.JAVA_SPRING: ["pom.xml", "build.gradle", "src/main/java"],
TechStack.NODE_EXPRESS: ["package.json", "app.js", "server.js"],
TechStack.NODE_NEST: ["nest-cli.json", "@nestjs"],
TechStack.PYTHON_DJANGO: ["manage.py", "django", "requirements.txt"],
TechStack.PYTHON_FASTAPI: ["fastapi", "main.py"],
TechStack.GO_GIN: ["go.mod", "gin"],
TechStack.GO_FIBER: ["go.mod", "fiber"],
TechStack.REACT: ["package.json", "react", "src/App.js", "src/App.tsx", "vite.config.ts", "vite.config.js"],
TechStack.VUE: ["package.json", "vue", "vue.config.js", "vite.config.ts", "vite.config.js"],
TechStack.ANGULAR: ["angular.json", "@angular"],
}
@classmethod
def detect(cls, project_dir: str) -> List[TechStack]:
"""检测项目技术栈"""
detected = []
project_path = Path(project_dir)
for tech, indicators in cls.DETECTION_RULES.items():
found = False
for indicator in indicators:
# 递归检查文件存在
for _ in project_path.rglob(indicator):
found = True
break
if found: break
# 递归检查文件内容
for file in project_path.rglob("*.json"):
try:
with open(file, 'r', encoding='utf-8') as f:
content = f.read()
if indicator in content:
found = True
break
except:
pass
if found: break
if found and tech not in detected:
detected.append(tech)
return detected if detected else [TechStack.UNKNOWN]
# 项目类型判断器
class ProjectTypeDetector:
"""项目类型检测器"""
@classmethod
def detect(cls, project_dir: str, tech_stacks: List[TechStack]) -> ProjectType:
"""检测项目类型"""
project_path = Path(project_dir)
# 检查桌面应用特征
desktop_indicators = ["electron", "qt", "swing", "wpf", ".desktop"]
for indicator in desktop_indicators:
if any(indicator in str(tech.value).lower() for tech in tech_stacks):
return ProjectType.DESKTOP
# 检查移动应用特征
mobile_indicators = ["react-native", "flutter", "android", "ios"]
for indicator in mobile_indicators:
if any(indicator in str(tech.value).lower() for tech in tech_stacks):
return ProjectType.MOBILE
# 检查Docker配置
if (project_path / "Dockerfile").exists() or (project_path / "docker-compose.yml").exists():
# 检查微服务特征
if (project_path / "services").exists() or len(list(project_path.glob("**/Dockerfile"))) > 1:
return ProjectType.DOCKER_MICROSERVICES
return ProjectType.DOCKER_WEB
# 检查API服务特征
if any(tech in [TechStack.NODE_EXPRESS, TechStack.NODE_NEST,
TechStack.PYTHON_FASTAPI, TechStack.GO_GIN] for tech in tech_stacks):
return ProjectType.DOCKER_API
return ProjectType.UNKNOWN
def main():
"""命令行入口"""
if len(sys.argv) < 2:
print("Usage: workflow_manager.py <command> [args...]")
print("Commands: init, status, resume, report")
sys.exit(1)
command = sys.argv[1]
if command == "init":
if len(sys.argv) < 5:
print("Usage: workflow_manager.py init <project_dir> <name> <git_url> [port_range]")
sys.exit(1)
project_dir = sys.argv[2]
name = sys.argv[3]
git_url = sys.argv[4]
port_range = sys.argv[5] if len(sys.argv) > 5 else None
manager = WorkflowManager(project_dir)
manager.init_project(name, git_url, port_range)
print(f"项目已初始化: {name}")
elif command == "start_step":
if len(sys.argv) < 4:
print("Usage: workflow_manager.py start_step <project_dir> <step_id>")
sys.exit(1)
project_dir = sys.argv[2]
step_id = int(sys.argv[3])
manager = WorkflowManager(project_dir)
if manager.load_state():
if manager.start_step(step_id):
print(f"步骤 {step_id} 已开始")
else:
print(f"步骤 {step_id} 开始失败")
else:
print("未找到状态文件")
elif command == "complete_step":
if len(sys.argv) < 4:
print("Usage: workflow_manager.py complete_step <project_dir> <step_id> [notes]")
sys.exit(1)
project_dir = sys.argv[2]
step_id = int(sys.argv[3])
notes = sys.argv[4] if len(sys.argv) > 4 else ""
manager = WorkflowManager(project_dir)
if manager.load_state():
manager.complete_step(step_id, notes)
print(f"步骤 {step_id} 已完成")
else:
print("未找到状态文件")
elif command == "detect":
if len(sys.argv) < 3:
print("Usage: workflow_manager.py detect <project_dir>")
sys.exit(1)
project_dir = sys.argv[2]
manager = WorkflowManager(project_dir)
if manager.load_state():
tech_stack = TechStackDetector.detect(project_dir)
project_type = ProjectTypeDetector.detect(project_dir, tech_stack)
manager.project_info.tech_stack = tech_stack
manager.project_info.project_type = project_type
manager.start_step(2)
manager.complete_step(2, f"Detected tech stack: {[ts.value for ts in tech_stack]}, project type: {project_type.value}")
print(f"Tech stack: {[ts.value for ts in tech_stack]}, Project type: {project_type.value}")
else:
print("未找到状态文件")
elif command == "status":
if len(sys.argv) < 3:
print("Usage: workflow_manager.py status <project_dir>")
sys.exit(1)
project_dir = sys.argv[2]
manager = WorkflowManager(project_dir)
if manager.load_state():
progress = manager.get_progress()
print(f"进度: {progress['completed']}/{progress['total']} ({progress['percentage']:.1f}%)")
current = manager.get_current_step()
if current:
print(f"当前步骤: 步骤 {current.id} - {current.name}")
else:
print("未找到状态文件")
elif command == "resume":
if len(sys.argv) < 3:
print("Usage: workflow_manager.py resume <project_dir>")
sys.exit(1)
project_dir = sys.argv[2]
manager = WorkflowManager(project_dir)
can_resume, step_id = manager.can_resume()
if can_resume:
print(f"可以从步骤 {step_id} 恢复执行")
else:
print("无法恢复,请重新初始化")
elif command == "report":
if len(sys.argv) < 3:
print("Usage: workflow_manager.py report <project_dir>")
sys.exit(1)
project_dir = sys.argv[2]
manager = WorkflowManager(project_dir)
if manager.load_state():
print(manager.generate_report())
else:
print("未找到状态文件")
if __name__ == "__main__":
main()