Skip to content

S
skills
Commit 4b9160b1 authored by 常佳涛's avatar 常佳涛
Browse files
Options

feat(vip-report): add CEP/VIS pipelines and clean repo ignore rules

parent aec72fb1
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 12098 additions and 59 deletions
.gitignore 0 → 100644
View file @ 4b9160b1
# Local tooling and IDE
.idea/
.codex/
# Python
__pycache__/
*.pyc
.venv/
.pydeps/
# Node
node_modules/
.npm-cache/
package-lock.json.*
# Browser automation caches
.playwright-cli/
.pw-daemon/
.pw-profile/
.chrome-cdp-test/
.chrome-user-data/
.mpl-cache/
# Generated output/artifacts
output/
*.zip
# Project-specific generated caches
vip-report/.playwright/profiles/
vip-report/output/
vip-report/node_modules/
vip-report/scripts/__pycache__/
vip-report/.venv/
vip-report/.playwright-cli/
vip-report/.pw-daemon/
vip-report/.pw-profile/
vip-report/.chrome-cdp-test/
vip-report/.chrome-user-data/
vip-report/.mpl-cache/
vip-report/.npm-cache/
vip-report/shadowbot_flow*_bundle/
vip-report/shadowbot_flow*_bundle.zip
import_cep_excel_to_mysql.py 0 → 100644
View file @ 4b9160b1
import argparse
import os
from pathlib import Path
import pandas as pd
import pymysql
def find_excel_file(pattern: str) -> str:
desktop = Path(r"C:\Users\ethan.chang\Desktop")
matches = sorted(desktop.glob(pattern), key=lambda p: p.stat().st_mtime, reverse=True)
if not matches:
raise FileNotFoundError("No matching Excel file found on Desktop.")
return str(matches[0])
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--excel", default=os.environ.get("EXCEL_PATH"))
parser.add_argument("--daily", required=True)
parser.add_argument("--table", default="cep_traffic_summary")
parser.add_argument("--pattern", default="*20260101-20260131*.xlsx")
args = parser.parse_args()
excel_path = args.excel or find_excel_file(args.pattern)
daily_val = args.daily
table = args.table
df = pd.read_excel(excel_path, sheet_name=0)
col_map = {
"流量类型": "traffic_type",
"一级入口": "primary_entry",
"二级入口": "secondary_entry",
"商详UV": "product_detail_uv",
"商详UV占比": "product_detail_uv_ratio",
"销售额(含拒退)": "sales_amount",
"销售额占比": "sales_ratio",
"销售量": "sales_volume",
"购买转化率": "purchase_conversion_rate",
"件单价": "unit_price",
"客户数": "customer_count",
"新客数": "new_customer_count",
"营销新客数": "marketing_new_customer_count",
"加购人数": "add_to_cart_users",
"商详UV价值": "product_detail_uv_value",
"ARPU": "arpu",
}
df = df.rename(columns=col_map)
pct_cols = ["product_detail_uv_ratio", "sales_ratio", "purchase_conversion_rate"]
for c in pct_cols:
df[c] = df[c].astype(str).str.replace("%", "", regex=False)
int_cols = [
"product_detail_uv",
"sales_volume",
"customer_count",
"new_customer_count",
"marketing_new_customer_count",
"add_to_cart_users",
]
for c in int_cols:
df[c] = pd.to_numeric(df[c], errors="coerce").fillna(0).astype(int)
num_cols = [
"product_detail_uv_ratio",
"sales_amount",
"sales_ratio",
"purchase_conversion_rate",
"unit_price",
"product_detail_uv_value",
"arpu",
]
for c in num_cols:
df[c] = pd.to_numeric(df[c], errors="coerce")
df["daily"] = daily_val
records = []
for _, r in df.iterrows():
records.append(
(
None if pd.isna(r["traffic_type"]) else str(r["traffic_type"]),
None if pd.isna(r["primary_entry"]) else str(r["primary_entry"]),
None if pd.isna(r["secondary_entry"]) else str(r["secondary_entry"]),
int(r["product_detail_uv"]),
None if pd.isna(r["product_detail_uv_ratio"]) else float(r["product_detail_uv_ratio"]),
None if pd.isna(r["sales_amount"]) else float(r["sales_amount"]),
None if pd.isna(r["sales_ratio"]) else float(r["sales_ratio"]),
int(r["sales_volume"]),
None if pd.isna(r["purchase_conversion_rate"]) else float(r["purchase_conversion_rate"]),
None if pd.isna(r["unit_price"]) else float(r["unit_price"]),
int(r["customer_count"]),
int(r["new_customer_count"]),
int(r["marketing_new_customer_count"]),
int(r["add_to_cart_users"]),
None if pd.isna(r["product_detail_uv_value"]) else float(r["product_detail_uv_value"]),
None if pd.isna(r["arpu"]) else float(r["arpu"]),
daily_val,
)
)
conn = pymysql.connect(
host="192.168.138.182",
port=3306,
user="ceptest",
password="ck@123456",
database="ckc_cep_db_test",
charset="utf8mb4",
autocommit=False,
)
old_table = "cep_traffic_summary_202601"
create_sql = f"""
CREATE TABLE IF NOT EXISTS `{table}` (
`id` BIGINT NOT NULL AUTO_INCREMENT,
`traffic_type` VARCHAR(64) NULL,
`primary_entry` VARCHAR(64) NULL,
`secondary_entry` VARCHAR(64) NULL,
`product_detail_uv` INT NULL,
`product_detail_uv_ratio` DECIMAL(7,2) NULL,
`sales_amount` DECIMAL(18,2) NULL,
`sales_ratio` DECIMAL(7,2) NULL,
`sales_volume` INT NULL,
`purchase_conversion_rate` DECIMAL(7,2) NULL,
`unit_price` DECIMAL(12,2) NULL,
`customer_count` INT NULL,
`new_customer_count` INT NULL,
`marketing_new_customer_count` INT NULL,
`add_to_cart_users` INT NULL,
`product_detail_uv_value` DECIMAL(12,2) NULL,
`arpu` DECIMAL(12,2) NULL,
`daily` CHAR(7) NOT NULL,
`created_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `idx_daily` (`daily`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""
insert_sql = f"""
INSERT INTO `{table}` (
traffic_type, primary_entry, secondary_entry,
product_detail_uv, product_detail_uv_ratio,
sales_amount, sales_ratio, sales_volume, purchase_conversion_rate,
unit_price, customer_count, new_customer_count,
marketing_new_customer_count, add_to_cart_users,
product_detail_uv_value, arpu, daily
) VALUES (
%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s
)
"""
try:
with conn.cursor() as cur:
cur.execute(f"SHOW TABLES LIKE '{old_table}'")
old_exists = cur.fetchone() is not None
cur.execute(f"SHOW TABLES LIKE '{table}'")
new_exists = cur.fetchone() is not None
if old_exists and not new_exists and old_table != table:
cur.execute(f"RENAME TABLE `{old_table}` TO `{table}`")
cur.execute(create_sql)
cur.execute(f"DELETE FROM `{table}` WHERE daily=%s", (daily_val,))
cur.executemany(insert_sql, records)
cur.execute(f"SELECT COUNT(*) FROM `{table}` WHERE daily=%s", (daily_val,))
count_daily = cur.fetchone()[0]
conn.commit()
print(f"TABLE={table}")
print(f"EXCEL={excel_path}")
print(f"INSERTED={len(records)}")
print(f"COUNT_DAILY={count_daily}")
finally:
conn.close()
if __name__ == "__main__":
main()
vip-report/.playwright/cli.config.json
View file @ 4b9160b1
{ {
"browser": { "browser": {
"launchOptions": { "launchOptions": {}
"args": ["--no-proxy-server"]
}
} }
} }
vip-report/Report.pptx
View file @ 4b9160b1
No preview for this file type
vip-report/bin/vip-report-cep-live-session-sync.ps1 0 → 100644
View file @ 4b9160b1
param(
[string]$Config = "D:\projects\skills\vip-report\config.yaml",
[string]$ReportMonth = "",
[int]$ReportYear = 0
)
$ErrorActionPreference = "Stop"
$root = Split-Path -Parent $PSScriptRoot
$args = @(
"$root\scripts\sync_cep_live_session_assets.py",
"--config", $Config
)
if ($ReportMonth -ne "") {
$args += @("--report-month", $ReportMonth)
}
if ($ReportYear -gt 0) {
$args += @("--report-year", [string]$ReportYear)
}
python @args
vip-report/bin/vip-report-cep-live-session-trend-sync.ps1 0 → 100644
View file @ 4b9160b1
param(
[string]$Config = "D:\projects\skills\vip-report\config.yaml",
[string]$ReportMonth = "",
[int]$ReportYear = 0
)
$ErrorActionPreference = "Stop"
$root = Split-Path -Parent $PSScriptRoot
$args = @(
"$root\scripts\sync_cep_live_session_trend_assets.py",
"--config", $Config
)
if ($ReportMonth -ne "") {
$args += @("--report-month", $ReportMonth)
}
if ($ReportYear -gt 0) {
$args += @("--report-year", [string]$ReportYear)
}
python @args
vip-report/bin/vip-report-cep-site-traffic-entry-sync.ps1 0 → 100644
View file @ 4b9160b1
param(
[string]$ConfigPath = "D:\projects\skills\vip-report\config.yaml",
[string]$ReportMonth = "",
[int]$ReportYear = 0,
[int]$CompareYear = 0
)
$ErrorActionPreference = "Stop"
$root = Split-Path -Parent $PSScriptRoot
$pythonArgs = @(
"$root\scripts\sync_cep_site_traffic_entry_assets.py",
"--config", "$ConfigPath"
)
if ($ReportMonth) {
$pythonArgs += @("--report-month", "$ReportMonth")
}
if ($ReportYear -gt 0) {
$pythonArgs += @("--report-year", "$ReportYear")
}
if ($CompareYear -gt 0) {
$pythonArgs += @("--compare-year", "$CompareYear")
}
python @pythonArgs
vip-report/bin/vip-report-cep-traffic-summary-sync.ps1 0 → 100644
View file @ 4b9160b1
param(
[string]$ConfigPath = "D:\projects\skills\vip-report\config.yaml",
[string]$ReportMonth = "",
[int]$ReportYear = 0,
[int]$CompareYear = 0
)
$ErrorActionPreference = "Stop"
$root = Split-Path -Parent $PSScriptRoot
$pythonArgs = @(
"$root\scripts\sync_cep_traffic_summary_assets.py",
"--config", "$ConfigPath"
)
if ($ReportMonth) {
$pythonArgs += @("--report-month", "$ReportMonth")
}
if ($ReportYear -gt 0) {
$pythonArgs += @("--report-year", "$ReportYear")
}
if ($CompareYear -gt 0) {
$pythonArgs += @("--compare-year", "$CompareYear")
}
python @pythonArgs
vip-report/bin/vip-report-cep-traffic-sync.ps1 0 → 100644
View file @ 4b9160b1
param(
[string]$ConfigPath = "D:\projects\skills\vip-report\config.yaml",
[string]$ReportMonth = "",
[int]$ReportYear = 0,
[int]$CompareYear = 0
)
$ErrorActionPreference = "Stop"
$root = Split-Path -Parent $PSScriptRoot
$pythonArgs = @(
"$root\scripts\sync_cep_traffic_assets.py",
"--config", "$ConfigPath"
)
if ($ReportMonth) {
$pythonArgs += @("--report-month", "$ReportMonth")
}
if ($ReportYear -gt 0) {
$pythonArgs += @("--report-year", "$ReportYear")
}
if ($CompareYear -gt 0) {
$pythonArgs += @("--compare-year", "$CompareYear")
}
python @pythonArgs
vip-report/bin/vip-report-render.ps1
View file @ 4b9160b1
...@@ -24,7 +24,7 @@ if (-not $TemplatePath) { ...@@ -24,7 +24,7 @@ if (-not $TemplatePath) {
} }
if (-not $OutputPath) { if (-not $OutputPath) {
$OutputPath = "output\generated-from-template.pptx" $OutputPath = $TemplatePath
} }
$TemplatePath = Resolve-ProjectPath $TemplatePath $TemplatePath = Resolve-ProjectPath $TemplatePath
......
vip-report/bin/vip-report-tableau-sync.ps1
View file @ 4b9160b1
...@@ -18,8 +18,6 @@ param( ...@@ -18,8 +18,6 @@ param(
$ErrorActionPreference = "Stop" $ErrorActionPreference = "Stop"
$root = Split-Path -Parent $PSScriptRoot $root = Split-Path -Parent $PSScriptRoot
$outputDir = if ($OutputDir) { [System.IO.Path]::GetFullPath($OutputDir) } else { Join-Path $root "output" }
$mergedOpsPath = Join-Path $outputDir "render-ops.tableau-selection.live.json"
$commandStartedAt = Get-Date $commandStartedAt = Get-Date
if (-not $ConfigPath) { if (-not $ConfigPath) {
...@@ -29,12 +27,27 @@ if (-not $TemplatePath) { ...@@ -29,12 +27,27 @@ if (-not $TemplatePath) {
$TemplatePath = Join-Path $root "Report.pptx" $TemplatePath = Join-Path $root "Report.pptx"
} }
if (-not $OutputPath) { if (-not $OutputPath) {
$OutputPath = Join-Path $outputDir "generated-tableau-selection.pptx" $OutputPath = $TemplatePath
} }
if (-not (Test-Path -LiteralPath $ConfigPath)) { if (-not (Test-Path -LiteralPath $ConfigPath)) {
throw "ConfigPath not found: $ConfigPath" throw "ConfigPath not found: $ConfigPath"
} }
function Resolve-ConfiguredWorkdir {
param([string]$ConfigPathValue)
$text = Get-Content -LiteralPath $ConfigPathValue -Encoding UTF8 -Raw
$match = [regex]::Match($text, "(?m)^\s{2}workdir:\s*[""']?(?<path>[^""'\r\n]+)[""']?\s*$")
if ($match.Success) {
return [System.IO.Path]::GetFullPath($match.Groups["path"].Value.Trim())
}
return Join-Path $root "output"
}
$outputDir = if ($OutputDir) { [System.IO.Path]::GetFullPath($OutputDir) } else { Resolve-ConfiguredWorkdir -ConfigPathValue $ConfigPath }
$mergedOpsPath = Join-Path $outputDir "render-ops.tableau-selection.live.json"
if (-not (Test-Path -LiteralPath $TemplatePath) -and -not $NoRender.IsPresent) { if (-not (Test-Path -LiteralPath $TemplatePath) -and -not $NoRender.IsPresent) {
throw "TemplatePath not found: $TemplatePath" throw "TemplatePath not found: $TemplatePath"
} }
...@@ -43,6 +56,10 @@ if (-not (Test-Path -LiteralPath $outputDir)) { ...@@ -43,6 +56,10 @@ if (-not (Test-Path -LiteralPath $outputDir)) {
} }
function Resolve-PythonCommand { function Resolve-PythonCommand {
$venvPython = Join-Path $root ".venv\Scripts\python.exe"
if (Test-Path -LiteralPath $venvPython) {
return ,$venvPython
}
if (Get-Command py -ErrorAction SilentlyContinue) { if (Get-Command py -ErrorAction SilentlyContinue) {
return @("py", "-3") return @("py", "-3")
} }
...@@ -52,7 +69,7 @@ function Resolve-PythonCommand { ...@@ -52,7 +69,7 @@ function Resolve-PythonCommand {
throw "Python launcher not found. Please install py or python." throw "Python launcher not found. Please install py or python."
} }
$pythonCommand = Resolve-PythonCommand $pythonCommand = @(Resolve-PythonCommand)
function Format-Elapsed { function Format-Elapsed {
param([TimeSpan]$Duration) param([TimeSpan]$Duration)
......
vip-report/bin/vip-report-uv-lfl-compose.ps1 0 → 100644
View file @ 4b9160b1
param(
[string]$TableImage = "D:\projects\skills\vip-report\output\vip-report\assets\cep-traffic\cep-traffic-summary.png",
[string]$LineImage = "D:\projects\skills\vip-report\output\vip-report\assets\cep-site-traffic-entry\cep-site-traffic-entry-lfl.png",
[string]$EntryImage = "D:\projects\skills\vip-report\output\vip-report\assets\cep-traffic-summary\cep-traffic-summary-lfl.png",
[string]$OutputPath = "D:\projects\skills\vip-report\output\vip-report\assets\uv-lfl\uv-lfl-report.png",
[int]$ContentWidth = 1040
)
$ErrorActionPreference = "Stop"
$root = Split-Path -Parent $PSScriptRoot
python "$root\scripts\compose_uv_lfl_report.py" `
--table-image "$TableImage" `
--line-image "$LineImage" `
--entry-image "$EntryImage" `
--output "$OutputPath" `
--content-width "$ContentWidth"
vip-report/bin/vip-report-vis-live-download.ps1 0 → 100644
View file @ 4b9160b1
param(
[string]$Config = "D:\projects\skills\vip-report\config.yaml",
[string]$DownloadDir = "C:\vipdata",
[string]$ReportMonth = "",
[int]$ReportYear = 0,
[string]$Session = "vip-report-monthly-sales"
)
$ErrorActionPreference = "Stop"
$root = Split-Path -Parent $PSScriptRoot
$args = @(
"$root\scripts\sync_vis_live_session_downloads.py",
"--config", $Config,
"--download-dir", $DownloadDir,
"--session", $Session
)
if ($ReportMonth -ne "") {
$args += @("--report-month", $ReportMonth)
}
if ($ReportYear -gt 0) {
$args += @("--report-year", [string]$ReportYear)
}
python @args
vip-report/bin/vip-report-vis-live-load.ps1 0 → 100644
View file @ 4b9160b1
param(
[string]$Config = "D:\projects\skills\vip-report\config.yaml"
)
$ErrorActionPreference = "Stop"
$root = Split-Path -Parent $PSScriptRoot
python "$root\scripts\load_vis_live_session_to_mysql.py" `
--config "$Config"
vip-report/bin/vip-report-vis-vendor-goods-effect-download.ps1 0 → 100644
View file @ 4b9160b1
param(
[string]$Config = "D:\projects\skills\vip-report\config.yaml",
[string]$DownloadDir = "C:\vipdata",
[string]$ReportMonth = "",
[int]$ReportYear = 0,
[string]$Session = "vip-report-monthly-sales"
)
$ErrorActionPreference = "Stop"
$root = Split-Path -Parent $PSScriptRoot
$args = @(
"$root\scripts\sync_vis_vendor_goods_effect_download.py",
"--config", $Config,
"--download-dir", $DownloadDir,
"--session", $Session
)
if ($ReportMonth -ne "") {
$args += @("--report-month", $ReportMonth)
}
if ($ReportYear -gt 0) {
$args += @("--report-year", [string]$ReportYear)
}
python @args
vip-report/config.yaml
View file @ 4b9160b1
...@@ -3,31 +3,26 @@ tableau: ...@@ -3,31 +3,26 @@ tableau:
username: "ec_user01" username: "ec_user01"
password: "1!Cu&euRo17b6yu@R40gr7" password: "1!Cu&euRo17b6yu@R40gr7"
vip:
login_endpoint: ""
enabled: false
mysql: mysql:
host: "erp.charleskeith.cn" host: "192.168.138.182"
port: 30004 port: 3306
database: "ckc_cep_db" username: "ceptest"
username: "allen" password: "ck@123456"
password: "wangjun@123" database: "ckc_cep_db_test"
paths: paths:
template_pptx: "./Report.pptx" template_pptx: "D:\\projects\\skills\\vip-report\\Report.pptx"
slide_source_map: "./output/slide-source-map.yaml" slide_source_map: "D:\\projects\\skills\\vip-report\\slide-source-map.yaml"
workdir: "./output" workdir: "D:\\projects\\skills\\vip-report\\output\\vip-report"
render: render:
mode: "com" mode: "com"
baseline_output: "./output/generated-baseline.pptx"
manifest_output: "./output/template-manifest.json"
inventory_output: "./output/shape-inventory.json"
compare_output: "./output/baseline-compare.json"
report: report:
# 月份支持输入中文月份（如：二月）或数字（如：2/02），脚本会统一处理。 month_cn: "3"
month_cn: "一月"
year: 2026 year: 2026
compare_year: 2025 compare_year: 2025
vip:
login_endpoint: ""
vip-report/package-lock.json 0 → 100644
View file @ 4b9160b1
{
"name": "vip-report",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"dependencies": {
"playwright-core": "^1.59.1"
}
},
"node_modules/playwright-core": {
"version": "1.59.1",
"resolved": "https://registry.npmjs.org/playwright-core/-/playwright-core-1.59.1.tgz",
"integrity": "sha512-HBV/RJg81z5BiiZ9yPzIiClYV/QMsDCKUyogwH9p3MCP6IYjUFu/MActgYAvK0oWyV9NlwM3GLBjADyWgydVyg==",
"license": "Apache-2.0",
"bin": {
"playwright-core": "cli.js"
},
"engines": {
"node": ">=18"
}
}
}
}
vip-report/package.json 0 → 100644
View file @ 4b9160b1
{
"dependencies": {
"playwright-core": "^1.59.1"
}
}
vip-report/scripts/compose_live_goods_effect_report.py 0 → 100644
View file @ 4b9160b1
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from typing import Any
SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPT_DIR))
for dep_dir in [SCRIPT_DIR / ".pydeps", SCRIPT_DIR.parent / ".pydeps", Path.cwd() / ".pydeps"]:
if dep_dir.exists() and str(dep_dir) not in sys.path:
sys.path.insert(0, str(dep_dir))
from PIL import Image, ImageDraw, ImageFont
import yaml
from runtime_config import add_config_source_args, load_runtime_config
def load_font(size: int, *, bold: bool = False) -> ImageFont.ImageFont:
candidates: list[str] = []
if bold:
candidates.extend([r"C:\Windows\Fonts\msyhbd.ttc", r"C:\Windows\Fonts\simhei.ttf", r"C:\Windows\Fonts\arialbd.ttf"])
candidates.extend([r"C:\Windows\Fonts\msyh.ttc", r"C:\Windows\Fonts\simsun.ttc", r"C:\Windows\Fonts\arial.ttf"])
for p in candidates:
try:
return ImageFont.truetype(p, size=size)
except OSError:
continue
return ImageFont.load_default()
def resize_to_width(img: Image.Image, width: int) -> Image.Image:
if img.width == width:
return img
h = max(1, int(img.height * width / img.width))
return img.resize((width, h), Image.Resampling.LANCZOS)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Compose goods-effect flow images into one report.")
add_config_source_args(parser, default_config=Path(r"D:\projects\skills\vip-report\config.yaml"))
parser.add_argument("--summary-image", default="")
parser.add_argument("--top5-image", default="")
parser.add_argument("--output", default="")
parser.add_argument("--content-width", type=int, default=0, help="0 = auto use max source width")
return parser.parse_args()
def defaults_from_config(cfg: dict[str, Any]) -> dict[str, Path]:
workdir = Path(cfg["paths"]["workdir"]).resolve()
base = workdir / "assets" / "cep-live-goods-effect"
return {
"summary": base / "cep-live-goods-effect-summary.png",
"top5": base / "cep-live-goods-effect-top5.png",
"output": base / "cep-live-goods-effect-combined.png",
}
def resolve(p: str, default: Path) -> Path:
return Path(p).resolve() if p else default
def main() -> None:
args = parse_args()
config_path = Path(args.config or r"D:\projects\skills\vip-report\config.yaml").resolve()
cfg = load_runtime_config(args, default_config=config_path)
defaults = defaults_from_config(cfg)
summary_path = resolve(args.summary_image, defaults["summary"])
top5_path = resolve(args.top5_image, defaults["top5"])
output_path = resolve(args.output, defaults["output"])
if not summary_path.exists():
raise FileNotFoundError(f"summary image not found: {summary_path}")
if not top5_path.exists():
raise FileNotFoundError(f"top5 image not found: {top5_path}")
summary_img = Image.open(summary_path).convert("RGB")
top5_img = Image.open(top5_path).convert("RGB")
content_w = args.content_width if args.content_width > 0 else max(summary_img.width, top5_img.width)
summary_img = resize_to_width(summary_img, content_w)
top5_img = resize_to_width(top5_img, content_w)
pad_x = 24
top_h = 56
gap = 10
bottom_pad = 16
canvas_w = content_w + pad_x * 2
canvas_h = top_h + summary_img.height + gap + top5_img.height + bottom_pad
canvas = Image.new("RGB", (canvas_w, canvas_h), "#e7e7e7")
draw = ImageDraw.Draw(canvas)
draw.text((pad_x, 12), "LIVE GOODS EFFECT", fill="#111111", font=load_font(34, bold=True))
draw.text((canvas_w - 200, 16), "CHARLES & KEITH", fill="#c9c9c9", font=load_font(24, bold=True))
y = top_h
canvas.paste(summary_img, (pad_x + (content_w - summary_img.width) // 2, y))
y += summary_img.height + gap
canvas.paste(top5_img, (pad_x + (content_w - top5_img.width) // 2, y))
output_path.parent.mkdir(parents=True, exist_ok=True)
canvas.save(output_path)
print(json.dumps({"ok": True, "image_path": str(output_path)}, ensure_ascii=False))
if __name__ == "__main__":
main()
vip-report/scripts/compose_live_session_report.py 0 → 100644
View file @ 4b9160b1
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPT_DIR))
for dep_dir in [SCRIPT_DIR / ".pydeps", SCRIPT_DIR.parent / ".pydeps", Path.cwd() / ".pydeps"]:
if dep_dir.exists() and str(dep_dir) not in sys.path:
sys.path.insert(0, str(dep_dir))
from PIL import Image, ImageDraw, ImageFont
import yaml
from runtime_config import add_config_source_args, load_runtime_config
def load_font(size: int, *, bold: bool = False) -> ImageFont.ImageFont:
candidates: list[str] = []
if bold:
candidates.extend([r"C:\Windows\Fonts\msyhbd.ttc", r"C:\Windows\Fonts\simhei.ttf", r"C:\Windows\Fonts\arialbd.ttf"])
candidates.extend([r"C:\Windows\Fonts\msyh.ttc", r"C:\Windows\Fonts\simsun.ttc", r"C:\Windows\Fonts\arial.ttf"])
for path in candidates:
try:
return ImageFont.truetype(path, size=size)
except OSError:
continue
return ImageFont.load_default()
def resize_to_width(img: Image.Image, width: int) -> Image.Image:
if img.width == width:
return img
target_h = int(img.height * width / img.width)
return img.resize((width, target_h), Image.Resampling.LANCZOS)
def scale_image(img: Image.Image, factor: float) -> Image.Image:
if abs(factor - 1.0) < 1e-6:
return img
w = max(1, int(img.width * factor))
h = max(1, int(img.height * factor))
# For chart/table text assets, NEAREST keeps stroke edges crisper than smooth interpolation.
return img.resize((w, h), Image.Resampling.NEAREST)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Compose livestream report image from 3 source images.")
add_config_source_args(parser, default_config=Path(r"D:\projects\skills\vip-report\config.yaml"))
parser.add_argument("--summary-image", default="")
parser.add_argument("--trend-image", default="")
parser.add_argument("--pivot-image", default="")
parser.add_argument("--output", default="")
parser.add_argument("--content-width", type=int, default=0, help="0 means keep source widths and center align.")
parser.add_argument("--summary-crop-top", type=int, default=52)
parser.add_argument("--summary-scale", type=float, default=1.8)
parser.add_argument("--trend-scale", type=float, default=1.3)
parser.add_argument("--pivot-scale", type=float, default=1.0)
return parser.parse_args()
def default_paths_from_config(config: dict) -> dict[str, Path]:
workdir = Path(config["paths"]["workdir"]).resolve()
assets = workdir / "assets"
return {
"summary": assets / "cep-live-session" / "cep-live-session-summary.png",
"trend": assets / "cep-live-session" / "cep-live-session-sales-penetration-trend.png",
"pivot": assets / "cep-live-goods-effect" / "cep-live-goods-effect-pivot.png",
"output": assets / "cep-live-session" / "cep-live-session-combined.png",
}
def resolve_path(arg_value: str, default: Path) -> Path:
return Path(arg_value).resolve() if arg_value else default
def main() -> None:
args = parse_args()
config_path = Path(args.config or r"D:\projects\skills\vip-report\config.yaml").resolve()
config = load_runtime_config(args, default_config=config_path)
defaults = default_paths_from_config(config)
summary_path = resolve_path(args.summary_image, defaults["summary"])
trend_path = resolve_path(args.trend_image, defaults["trend"])
pivot_path = resolve_path(args.pivot_image, defaults["pivot"])
output_path = resolve_path(args.output, defaults["output"])
for p in [summary_path, trend_path, pivot_path]:
if not p.exists():
raise FileNotFoundError(f"missing source image: {p}")
summary_img = Image.open(summary_path).convert("RGB")
if args.summary_crop_top > 0 and args.summary_crop_top < summary_img.height:
summary_img = summary_img.crop((0, args.summary_crop_top, summary_img.width, summary_img.height))
trend_img = Image.open(trend_path).convert("RGB")
pivot_img = Image.open(pivot_path).convert("RGB")
summary_img = scale_image(summary_img, args.summary_scale)
trend_img = scale_image(trend_img, args.trend_scale)
pivot_img = scale_image(pivot_img, args.pivot_scale)
if args.content_width > 0:
summary_img = resize_to_width(summary_img, args.content_width)
trend_img = resize_to_width(trend_img, args.content_width)
pivot_img = resize_to_width(pivot_img, args.content_width)
pad_x = 34
top_h = 56
gap = 10
bottom_pad = 18
content_w = max(summary_img.width, trend_img.width, pivot_img.width)
canvas_w = content_w + pad_x * 2
canvas_h = top_h + summary_img.height + gap + trend_img.height + gap + pivot_img.height + bottom_pad
canvas = Image.new("RGB", (canvas_w, canvas_h), "#e7e7e7")
draw = ImageDraw.Draw(canvas)
title_font = load_font(36, bold=True)
brand_font = load_font(24, bold=True)
draw.text((pad_x, 14), "LIVESTREAM", fill="#111111", font=title_font)
draw.text((canvas_w - 176, 18), "CHARLES & KEITH", fill="#c9c9c9", font=brand_font)
y = top_h
canvas.paste(summary_img, (pad_x + (content_w - summary_img.width) // 2, y))
y += summary_img.height + gap
canvas.paste(trend_img, (pad_x + (content_w - trend_img.width) // 2, y))
y += trend_img.height + gap
canvas.paste(pivot_img, (pad_x + (content_w - pivot_img.width) // 2, y))
output_path.parent.mkdir(parents=True, exist_ok=True)
canvas.save(output_path)
print(json.dumps({"ok": True, "image_path": str(output_path)}, ensure_ascii=False))
if __name__ == "__main__":
main()
vip-report/scripts/compose_member_overview_report.py 0 → 100644
View file @ 4b9160b1
from __future__ import annotations
import argparse
import re
import json
import sys
from pathlib import Path
from typing import Any
import yaml
SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPT_DIR))
for dep_dir in [SCRIPT_DIR / ".pydeps", SCRIPT_DIR.parent / ".pydeps", Path.cwd() / ".pydeps"]:
if dep_dir.exists() and str(dep_dir) not in sys.path:
sys.path.insert(0, str(dep_dir))
from PIL import Image, ImageDraw, ImageFont
from runtime_config import add_config_source_args, load_runtime_config
def load_font(size: int, *, bold: bool = False):
candidates: list[str] = []
if bold:
candidates.extend([r"C:\Windows\Fonts\msyhbd.ttc", r"C:\Windows\Fonts\simhei.ttf", r"C:\Windows\Fonts\arialbd.ttf"])
candidates.extend([r"C:\Windows\Fonts\msyh.ttc", r"C:\Windows\Fonts\simsun.ttc", r"C:\Windows\Fonts\arial.ttf"])
for path in candidates:
try:
return ImageFont.truetype(path, size=size)
except OSError:
continue
return ImageFont.load_default()
def resize_to_width(img: Image.Image, width: int) -> Image.Image:
if img.width == width:
return img
h = int(img.height * width / img.width)
return img.resize((width, h), Image.Resampling.LANCZOS)
def resize_to_height(img: Image.Image, height: int) -> Image.Image:
if img.height == height:
return img
w = int(img.width * height / img.height)
return img.resize((w, height), Image.Resampling.LANCZOS)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Compose membership report image from 4 source images.")
add_config_source_args(parser, default_config=Path(r"D:\projects\skills\vip-report\config.yaml"))
parser.add_argument("--metrics-image", default="")
parser.add_argument("--conversion-image", default="")
parser.add_argument("--order-image", default="")
parser.add_argument("--channel-image", default="")
parser.add_argument("--output", default="")
parser.add_argument("--content-width", type=int, default=1060)
parser.add_argument("--metrics-crop-top", type=int, default=0)
return parser.parse_args()
def parse_month_number(raw: Any) -> int:
if raw is None:
return 1
if isinstance(raw, int):
return min(12, max(1, raw))
text = str(raw).strip()
m = re.search(r"(1[0-2]|0?[1-9])", text)
if m:
return int(m.group(1))
return 1
def default_paths_from_config(config: dict[str, Any]) -> dict[str, Path]:
workdir = Path(config["paths"]["workdir"]).resolve()
assets = workdir / "assets" / "cep-member-overview"
report_cfg = config.get("report", {})
year = int(report_cfg.get("year", 2026))
month = parse_month_number(report_cfg.get("month_cn", "1"))
ym = f"{year}{month:02d}"
return {
"metrics": assets / f"member-overview-metrics-{ym}.png",
"conversion": assets / f"member-consumption-convert-{ym}.png",
"order": assets / f"member-order-distribution-{ym}.png",
"channel": assets / f"member-channel-trend-{ym}.png",
"output": assets / f"member-overview-combined-{ym}.png",
}
def resolve_path(arg_value: str, default: Path) -> Path:
return Path(arg_value).resolve() if arg_value else default
def main() -> None:
args = parse_args()
config_path = Path(args.config or r"D:\projects\skills\vip-report\config.yaml").resolve()
cfg = load_runtime_config(args, default_config=config_path)
defaults = default_paths_from_config(cfg)
metrics_path = resolve_path(args.metrics_image, defaults["metrics"])
conversion_path = resolve_path(args.conversion_image, defaults["conversion"])
order_path = resolve_path(args.order_image, defaults["order"])
channel_path = resolve_path(args.channel_image, defaults["channel"])
output_path = resolve_path(args.output, defaults["output"])
for p in [metrics_path, conversion_path, order_path, channel_path]:
if not p.exists():
raise FileNotFoundError(f"missing source image: {p}")
metrics_img = Image.open(metrics_path).convert("RGB")
if 0 < args.metrics_crop_top < metrics_img.height:
metrics_img = metrics_img.crop((0, args.metrics_crop_top, metrics_img.width, metrics_img.height))
conversion_img = Image.open(conversion_path).convert("RGB")
order_img = Image.open(order_path).convert("RGB")
channel_img = Image.open(channel_path).convert("RGB")
metrics_img = resize_to_width(metrics_img, args.content_width)
conversion_img = resize_to_width(conversion_img, args.content_width)
bottom_gap = 12
left_target_w = int((args.content_width - bottom_gap) * 0.36)
right_target_w = args.content_width - bottom_gap - left_target_w
order_img = resize_to_width(order_img, left_target_w)
channel_img = resize_to_width(channel_img, right_target_w)
target_bottom_h = min(order_img.height, channel_img.height)
order_img = resize_to_height(order_img, target_bottom_h)
channel_img = resize_to_height(channel_img, target_bottom_h)
pad_x = 34
top_h = 56
gap = 12
bottom_pad = 20
canvas_w = args.content_width + pad_x * 2
canvas_h = top_h + metrics_img.height + gap + conversion_img.height + gap + target_bottom_h + bottom_pad
canvas = Image.new("RGB", (canvas_w, canvas_h), "#e7e7e7")
draw = ImageDraw.Draw(canvas)
title_font = load_font(36, bold=True)
brand_font = load_font(24, bold=True)
draw.text((pad_x, 14), "MEMBERSHIP", fill="#111111", font=title_font)
draw.text((canvas_w - 180, 18), "CHARLES & KEITH", fill="#c9c9c9", font=brand_font)
y = top_h
canvas.paste(metrics_img, (pad_x, y))
y += metrics_img.height + gap
canvas.paste(conversion_img, (pad_x, y))
y += conversion_img.height + gap
canvas.paste(order_img, (pad_x, y))
canvas.paste(channel_img, (pad_x + order_img.width + bottom_gap, y))
output_path.parent.mkdir(parents=True, exist_ok=True)
canvas.save(output_path)
print(json.dumps({"ok": True, "image_path": str(output_path)}, ensure_ascii=False))
if __name__ == "__main__":
main()
vip-report/scripts/compose_rejection_report.py 0 → 100644
View file @ 4b9160b1
from __future__ import annotations
import argparse
import re
from pathlib import Path
from typing import Any
from PIL import Image, ImageDraw, ImageFont
import yaml
def load_font(size: int, *, bold: bool = False) -> ImageFont.ImageFont:
candidates: list[str] = []
if bold:
candidates.extend([r"C:\Windows\Fonts\arialbd.ttf", r"C:\Windows\Fonts\msyhbd.ttc", r"C:\Windows\Fonts\simhei.ttf"])
candidates.extend([r"C:\Windows\Fonts\arial.ttf", r"C:\Windows\Fonts\msyh.ttc", r"C:\Windows\Fonts\simsun.ttc"])
for p in candidates:
try:
return ImageFont.truetype(p, size=size)
except OSError:
continue
return ImageFont.load_default()
def resize_to_width(img: Image.Image, width: int) -> Image.Image:
if img.width == width:
return img
h = max(1, int(img.height * width / img.width))
return img.resize((width, h), Image.Resampling.LANCZOS)
def resize_to_height(img: Image.Image, height: int) -> Image.Image:
if img.height == height:
return img
w = max(1, int(img.width * height / img.height))
return img.resize((w, height), Image.Resampling.LANCZOS)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Compose RETURN report image from three source images.")
parser.add_argument("--config", default=r"D:\projects\skills\vip-report\config.yaml")
parser.add_argument("--core-image", default="")
parser.add_argument("--lfl-image", default="")
parser.add_argument("--top-image", default="")
parser.add_argument("--output", default="")
parser.add_argument("--content-width", type=int, default=1088)
parser.add_argument("--bottom-left-ratio", type=float, default=0.42)
parser.add_argument("--core-crop-bottom", type=int, default=185)
return parser.parse_args()
def parse_month_number(raw: Any) -> int:
if raw is None:
return 1
if isinstance(raw, int):
return min(12, max(1, raw))
text = str(raw).strip()
m = re.search(r"(1[0-2]|0?[1-9])", text)
if m:
return int(m.group(1))
return 1
def defaults_from_config(config_path: Path) -> dict[str, Path]:
cfg = yaml.safe_load(config_path.read_text(encoding="utf-8"))
workdir = Path(cfg["paths"]["workdir"]).resolve()
base = workdir / "assets" / "cep-rejection-analysis"
report_cfg = cfg.get("report", {})
year = int(report_cfg.get("year", 2026))
month = parse_month_number(report_cfg.get("month_cn", "1"))
ym = f"{year}{month:02d}"
return {
"core": base / f"rejection-analysis-core-{ym}.png",
"lfl": base / f"cep-rejection-lfl-{ym}.png",
"top": base / "cep-rejection-goods-top5.png",
"output": base / f"cep-rejection-combined-{ym}.png",
}
def resolve(p: str, default: Path) -> Path:
return Path(p).resolve() if p else default
def main() -> None:
args = parse_args()
defaults = defaults_from_config(Path(args.config).resolve())
core_path = resolve(args.core_image, defaults["core"])
lfl_path = resolve(args.lfl_image, defaults["lfl"])
top_path = resolve(args.top_image, defaults["top"])
output_path = resolve(args.output, defaults["output"])
for p in [core_path, lfl_path, top_path]:
if not p.exists():
raise FileNotFoundError(f"missing source image: {p}")
core_img = Image.open(core_path).convert("RGB")
if args.core_crop_bottom > 0 and args.core_crop_bottom < core_img.height:
core_img = core_img.crop((0, 0, core_img.width, core_img.height - args.core_crop_bottom))
lfl_img = Image.open(lfl_path).convert("RGB")
top_img = Image.open(top_path).convert("RGB")
content_w = args.content_width
core_img = resize_to_width(core_img, content_w)
gap = 12
left_w = int((content_w - gap) * args.bottom_left_ratio)
right_w = content_w - gap - left_w
lfl_img = resize_to_width(lfl_img, left_w)
top_img = resize_to_width(top_img, right_w)
bottom_h = min(lfl_img.height, top_img.height)
lfl_img = resize_to_height(lfl_img, bottom_h)
top_img = resize_to_height(top_img, bottom_h)
pad_x = 18
top_h = 52
row_gap = 12
bottom_pad = 16
canvas_w = content_w + pad_x * 2
canvas_h = top_h + core_img.height + row_gap + bottom_h + bottom_pad
canvas = Image.new("RGB", (canvas_w, canvas_h), "#e7e7e7")
draw = ImageDraw.Draw(canvas)
draw.text((pad_x + 10, 12), "RETURN", fill="#111111", font=load_font(36, bold=True))
draw.text((canvas_w - 220, 16), "CHARLES & KEITH", fill="#c9c9c9", font=load_font(24, bold=True))
y = top_h
canvas.paste(core_img, (pad_x, y))
y += core_img.height + row_gap
canvas.paste(lfl_img, (pad_x, y))
canvas.paste(top_img, (pad_x + left_w + gap, y))
output_path.parent.mkdir(parents=True, exist_ok=True)
canvas.save(output_path)
print(str(output_path))
if __name__ == "__main__":
main()
vip-report/scripts/compose_uv_lfl_report.py 0 → 100644
View file @ 4b9160b1
from __future__ import annotations
import argparse
from pathlib import Path
from PIL import Image, ImageDraw, ImageFont
def load_font(size: int, *, bold: bool = False):
candidates: list[str] = []
if bold:
candidates.extend([r"C:\Windows\Fonts\msyhbd.ttc", r"C:\Windows\Fonts\simhei.ttf"])
candidates.extend([r"C:\Windows\Fonts\msyh.ttc", r"C:\Windows\Fonts\simsun.ttc", r"C:\Windows\Fonts\arial.ttf"])
for path in candidates:
try:
return ImageFont.truetype(path, size=size)
except OSError:
continue
return ImageFont.load_default()
def resize_to_width(img: Image.Image, width: int) -> Image.Image:
if img.width == width:
return img
h = int(img.height * width / img.width)
return img.resize((width, h), Image.Resampling.LANCZOS)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Compose UV/LFL report image from 3 source images.")
parser.add_argument("--table-image", required=True)
parser.add_argument("--line-image", required=True)
parser.add_argument("--entry-image", required=True)
parser.add_argument("--output", required=True)
parser.add_argument("--content-width", type=int, default=1040)
return parser.parse_args()
def main() -> None:
args = parse_args()
table_img = resize_to_width(Image.open(args.table_image).convert("RGB"), args.content_width)
line_img = resize_to_width(Image.open(args.line_image).convert("RGB"), args.content_width)
entry_img = resize_to_width(Image.open(args.entry_image).convert("RGB"), args.content_width)
pad_x = 34
top_h = 56
gap = 12
bottom_pad = 24
canvas_w = args.content_width + pad_x * 2
canvas_h = top_h + table_img.height + gap + line_img.height + gap + entry_img.height + bottom_pad
canvas = Image.new("RGB", (canvas_w, canvas_h), "#e7e7e7")
draw = ImageDraw.Draw(canvas)
title_font = load_font(36, bold=True)
brand_font = load_font(24, bold=True)
draw.text((pad_x, 14), "UV LFL", fill="#111111", font=title_font)
draw.text((canvas_w - 210, 18), "CHARLES & KEITH", fill="#c9c9c9", font=brand_font)
y = top_h
canvas.paste(table_img, (pad_x, y))
y += table_img.height + gap
canvas.paste(line_img, (pad_x, y))
y += line_img.height + gap
canvas.paste(entry_img, (pad_x, y))
output_path = Path(args.output)
output_path.parent.mkdir(parents=True, exist_ok=True)
canvas.save(output_path)
print(str(output_path))
if __name__ == "__main__":
main()
vip-report/scripts/debug_compass_rejection_dom.py 0 → 100644
View file @ 4b9160b1
from __future__ import annotations
import json
import os
import shutil
import subprocess
import time
from pathlib import Path
import yaml
NPX_EXECUTABLE = shutil.which("npx.cmd") or shutil.which("npx") or "npx"
PLAYWRIGHT_CMD = [NPX_EXECUTABLE, "--yes", "--package", "@playwright/cli", "playwright-cli"]
SESSION_NAME = "vip-report-monthly-sales"
TARGET_URL = "https://compass.vip.com/frontend/index.html#/rejectionAnalysis/index"
def run_cmd(args: list[str], *, cwd: Path, timeout: int = 300) -> subprocess.CompletedProcess[str]:
env = os.environ.copy()
npm_cache = cwd / ".npm-cache"
npm_cache.mkdir(parents=True, exist_ok=True)
env["NPM_CONFIG_CACHE"] = str(npm_cache)
return subprocess.run(args, cwd=str(cwd), text=True, encoding="utf-8", errors="replace", capture_output=True, timeout=timeout, check=True, env=env)
def main() -> None:
cfg = yaml.safe_load(Path("config.yaml").read_text(encoding="utf-8"))
workdir = Path(cfg["paths"]["workdir"]).resolve()
js_path = workdir / f"tmp-debug-rejection-dom-{time.time_ns()}.js"
js_path.write_text(
"""async function(page) {
const sleep = (ms) => page.waitForTimeout(ms);
await page.goto('""" + TARGET_URL + """?t=' + Date.now(), { waitUntil: 'domcontentloaded', timeout: 120000 });
await sleep(9000);
const out = await page.evaluate(() => {
const visible = (el) => {
if (!el) return false;
const s = getComputedStyle(el);
const r = el.getBoundingClientRect();
return s.display !== 'none' && s.visibility !== 'hidden' && r.width > 1 && r.height > 1;
};
const compact = (s) => String(s || '').replace(/\\s+/g, ' ').trim().slice(0, 260);
const nodeInfo = (el) => {
let cur = el;
const chain = [];
for (let i = 0; cur && i < 6; i += 1, cur = cur.parentElement) {
chain.push({ tag: cur.tagName, cls: cur.className, text: compact(cur.textContent) });
}
const inputs = Array.from(el.querySelectorAll('input')).map((x) => ({ placeholder: x.getAttribute('placeholder'), value: x.value, cls: x.className }));
const r = el.getBoundingClientRect();
return { cls: el.className, text: compact(el.textContent), inputs, rect: { x: r.x, y: r.y, w: r.width, h: r.height }, chain };
};
const editors = Array.from(document.querySelectorAll('.el-date-editor,.el-date-editor--daterange,.el-date-editor--monthrange,.el-date-editor--month')).filter(visible).map(nodeInfo);
const forms = Array.from(document.querySelectorAll('.el-form-item,.search-item,.filter-item')).filter(visible).map(nodeInfo);
const buttons = Array.from(document.querySelectorAll('button,span')).filter((el) => visible(el) && compact(el.textContent).includes('查询')).map(nodeInfo);
const vueChain = [];
const firstEditor = Array.from(document.querySelectorAll('.el-date-editor--daterange')).filter(visible)[0];
let cur = firstEditor;
for (let i = 0; cur && i < 8; i += 1, cur = cur.parentElement) {
const vm = cur.__vue__;
vueChain.push({
tag: cur.tagName,
cls: cur.className,
text: compact(cur.textContent),
hasVue: !!vm,
vueKeys: vm ? Object.keys(vm).filter((k) => !k.startsWith('_')).slice(0, 80) : [],
dataKeys: vm && vm.$data ? Object.keys(vm.$data).slice(0, 80) : [],
props: vm && vm.$props ? Object.keys(vm.$props).slice(0, 80) : []
});
}
const parentChain = [];
const selector = Array.from(document.querySelectorAll('.gd-date-selector.date-selector')).filter((el) => visible(el) && !el.classList.contains('compare'))[0];
let vm = selector && selector.__vue__;
for (let i = 0; vm && i < 10; i += 1, vm = vm.$parent) {
const keys = Object.keys(vm).filter((k) => !k.startsWith('_'));
parentChain.push({
name: vm.$options && vm.$options.name,
tag: vm.$el && vm.$el.tagName,
cls: vm.$el && vm.$el.className,
text: vm.$el ? compact(vm.$el.textContent) : '',
keys: keys.slice(0, 160),
interesting: keys.filter((k) => /query|search|fetch|get|load|change|date|init|refresh|data/i.test(k)).slice(0, 120),
methodHints: Object.fromEntries(keys.filter((k) => /changeDateHandler|handlerConditionChange|getConfigs|getParams|handleUpdateOpts/.test(k)).map((k) => [k, String(vm[k]).slice(0, 500)])),
data: vm.$data ? Object.fromEntries(Object.entries(vm.$data).filter(([k]) => /date|time|query|search|param|condition|filter/i.test(k)).slice(0, 40)) : {}
});
}
return { editors, forms, buttons, vueChain, parentChain, bodyHead: compact(document.body.innerText) };
});
return out;
}
""",
encoding="utf-8",
)
try:
out = run_cmd(PLAYWRIGHT_CMD + ["--session", SESSION_NAME, "run-code", "--filename", str(js_path)], cwd=workdir, timeout=420).stdout
finally:
js_path.unlink(missing_ok=True)
for line in out.splitlines():
s = line.strip()
if s.startswith("{") and s.endswith("}"):
print(json.dumps(json.loads(s), ensure_ascii=False, indent=2))
return
print(out)
if __name__ == "__main__":
main()
vip-report/scripts/download_compass_ready_file.py 0 → 100644
View file @ 4b9160b1
from __future__ import annotations
import argparse
import json
import os
import shutil
import subprocess
import time
from pathlib import Path
from typing import Any
import yaml
NPX_EXECUTABLE = shutil.which("npx.cmd") or shutil.which("npx") or "npx"
PLAYWRIGHT_CMD = [NPX_EXECUTABLE, "--yes", "--package", "@playwright/cli", "playwright-cli"]
SESSION_NAME = "vip-report-monthly-sales"
DOWNLOAD_CENTER_URL = "https://compass.vip.com/frontend/index.html#/dataCapturing/downloadCenter"
DEFAULT_DOWNLOAD_DIR = r"C:\vipdata"
def run_cmd(args: list[str], *, cwd: Path, timeout: int = 300) -> subprocess.CompletedProcess[str]:
env = os.environ.copy()
npm_cache = cwd / ".npm-cache"
npm_cache.mkdir(parents=True, exist_ok=True)
env["NPM_CONFIG_CACHE"] = str(npm_cache)
return subprocess.run(
args,
cwd=str(cwd),
text=True,
encoding="utf-8",
errors="replace",
capture_output=True,
timeout=timeout,
check=True,
env=env,
)
def run_playwright(args: list[str], *, cwd: Path, timeout: int = 300) -> str:
return run_cmd(PLAYWRIGHT_CMD + args, cwd=cwd, timeout=timeout).stdout
def write_js(workdir: Path, name: str, content: str) -> Path:
p = workdir / name
p.write_text(content, encoding="utf-8")
return p
def run_code(session: str, script_path: Path, *, cwd: Path, timeout: int = 420) -> str:
return run_playwright(["--session", session, "run-code", "--filename", str(script_path)], cwd=cwd, timeout=timeout)
def build_js(spec: dict[str, Any]) -> str:
payload = json.dumps(spec, ensure_ascii=False)
return f"""async function(page) {{
const spec = {payload};
const sleep = (ms) => page.waitForTimeout(ms);
const safeName = (name) => String(name || '')
.replace(/[\\\\/:*?"<>|]/g, '_')
.replace(/\\s+/g, '_')
.replace(/_+/g, '_');
await page.goto(spec.url, {{ waitUntil: 'domcontentloaded', timeout: 120000 }});
await sleep(3500);
const rowsSeen = [];
for (let attempt = 0; attempt < spec.max_retries; attempt++) {{
const queryBtn = page.locator('button:has-text("查询"), span:has-text("查询"), button:has-text("鏌ヨ"), span:has-text("鏌ヨ")').first();
if (await queryBtn.count().catch(() => 0)) {{
await queryBtn.click({{ timeout: 2500 }}).catch(() => null);
}}
await sleep(2200);
const rows = await page.locator('tbody tr').count().catch(() => 0);
for (let i = 0; i < Math.min(rows, spec.scan_rows); i++) {{
const row = page.locator('tbody tr').nth(i);
const text = (await row.innerText().catch(() => '')).replace(/\\s+/g, ' ').trim();
const nameEl = row.locator('.status-name-span').first();
let fileName = await nameEl.getAttribute('title').catch(() => '');
if (!fileName) fileName = (await nameEl.innerText().catch(() => '')).trim();
if (!fileName) fileName = text;
rowsSeen.push({{ index: i + 1, fileName, text }});
const hasKeyword = spec.keywords.every((k) => String(fileName || text).includes(k));
const isSuccess = /执行成功|成功|鎵ц鎴愬姛|鎴愬姛/.test(text);
if (!hasKeyword || !isSuccess) continue;
const downloadBtn = row.locator('button:has-text("下载"), span:has-text("下载"), button:has-text("涓嬭浇"), span:has-text("涓嬭浇")').first();
if (!(await downloadBtn.count().catch(() => 0))) continue;
await page.locator('.el-loading-mask').waitFor({{ state: 'hidden', timeout: 10000 }}).catch(() => null);
const [dl] = await Promise.all([
page.waitForEvent('download', {{ timeout: 120000 }}),
downloadBtn.click({{ timeout: 10000, force: true }}),
]);
const suggested = dl.suggestedFilename();
const finalName = safeName(`${{spec.prefix}}_${{fileName || suggested}}`);
const targetPath = `${{spec.download_dir.replace(/\\\\/g, '/')}}/${{finalName}}`;
await dl.saveAs(targetPath);
return {{ ok: true, savedAs: targetPath, fileName, suggestedFilename: suggested, rowText: text }};
}}
}}
return {{ ok: false, code: 'READY_FILE_NOT_FOUND', keywords: spec.keywords, rowsSeen: rowsSeen.slice(0, 20), url: page.url() }};
}}
"""
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Download an already-ready Compass download-center file.")
p.add_argument("--config", default=r"D:\projects\skills\vip-report\config.yaml")
p.add_argument("--session", default=SESSION_NAME)
p.add_argument("--download-dir", default=DEFAULT_DOWNLOAD_DIR)
p.add_argument("--prefix", required=True)
p.add_argument("--keyword", action="append", default=[])
p.add_argument("--max-retries", type=int, default=5)
p.add_argument("--scan-rows", type=int, default=10)
return p.parse_args()
def main() -> None:
args = parse_args()
config_path = Path(args.config).resolve()
cfg = yaml.safe_load(config_path.read_text(encoding="utf-8"))
workdir = Path(cfg["paths"]["workdir"]).resolve()
download_dir = Path(args.download_dir).resolve()
download_dir.mkdir(parents=True, exist_ok=True)
js_path = write_js(
workdir,
f"tmp-download-ready-{args.prefix}-{time.time_ns()}.js",
build_js(
{
"url": DOWNLOAD_CENTER_URL,
"download_dir": str(download_dir),
"prefix": args.prefix,
"keywords": args.keyword,
"max_retries": args.max_retries,
"scan_rows": args.scan_rows,
}
),
)
try:
out = run_code(args.session, js_path, cwd=workdir, timeout=420)
finally:
js_path.unlink(missing_ok=True)
if "### Error" in out:
raise RuntimeError(out.strip())
json_line = ""
for line in out.splitlines():
s = line.strip()
if s.startswith("{") and s.endswith("}"):
json_line = s
break
if not json_line:
raise RuntimeError(f"No JSON result from run-code: {out}")
result = json.loads(json_line)
if not result.get("ok", False):
raise RuntimeError(json.dumps(result, ensure_ascii=False))
print(json.dumps(result, ensure_ascii=False))
if __name__ == "__main__":
main()
vip-report/scripts/load_compass_rejection_goods_cn_to_mysql.py 0 → 100644
View file @ 4b9160b1
from __future__ import annotations
import argparse
import re
import sys
from pathlib import Path
from typing import Any
DEPS_DIR = Path(__file__).resolve().parents[2] / ".pydeps"
if DEPS_DIR.exists():
sys.path.insert(0, str(DEPS_DIR))
import pandas as pd
import pymysql
import yaml
TABLE_NAME = "cep_rejection_goods_analysis"
def parse_month_number(raw: Any) -> int:
if raw is None:
return 1
if isinstance(raw, int):
return min(12, max(1, raw))
text = str(raw).strip()
m = re.search(r"(1[0-2]|0?[1-9])", text)
if m:
return int(m.group(1))
return 1
def resolve_target_month(config_path: Path, report_year_arg: int, report_month_arg: str) -> str:
cfg = yaml.safe_load(config_path.read_text(encoding="utf-8"))
report_cfg = cfg.get("report", {})
year = int(report_year_arg or report_cfg.get("year", 2026))
month = parse_month_number(report_month_arg or report_cfg.get("month_cn", "1"))
return f"{year}-{month:02d}"
def to_int(value: Any) -> int | None:
if value is None or (isinstance(value, float) and pd.isna(value)):
return None
s = str(value).strip().replace(",", "")
if not s:
return None
try:
return int(float(s))
except Exception:
return None
def to_float(value: Any) -> float | None:
if value is None or (isinstance(value, float) and pd.isna(value)):
return None
s = str(value).strip().replace(",", "").replace("%", "")
if not s:
return None
try:
return float(s)
except Exception:
return None
def to_str(value: Any) -> str | None:
if value is None or (isinstance(value, float) and pd.isna(value)):
return None
s = str(value).strip()
return s if s else None
def create_sql() -> str:
return f"""
CREATE TABLE IF NOT EXISTS `{TABLE_NAME}` (
`id` BIGINT NOT NULL AUTO_INCREMENT,
`data_month` CHAR(7) NOT NULL,
`time_range` VARCHAR(32) NULL,
`spu` VARCHAR(64) NULL,
`reason_level_1` VARCHAR(128) NULL,
`reason_level_2` VARCHAR(128) NULL,
`reason_issue_category` VARCHAR(128) NULL,
`reject_scene_level_1` VARCHAR(128) NULL,
`reject_scene_level_2` VARCHAR(128) NULL,
`product_name` VARCHAR(512) NULL,
`category_level_1` VARCHAR(128) NULL,
`category_level_2` VARCHAR(128) NULL,
`category_level_3` VARCHAR(128) NULL,
`article` VARCHAR(64) NULL,
`brand_sn` VARCHAR(64) NULL,
`brand_name` VARCHAR(128) NULL,
`min_price` DECIMAL(18,2) NULL,
`sales_qty` INT NULL,
`reject_qty` INT NULL,
`reject_amount` DECIMAL(18,2) NULL,
`reject_rate_qty` DECIMAL(18,6) NULL,
`reject_rate_amount` DECIMAL(18,6) NULL,
`source_file` VARCHAR(255) NOT NULL,
`created_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `idx_data_month` (`data_month`),
KEY `idx_article` (`article`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""
INSERT_SQL = f"""
INSERT INTO `{TABLE_NAME}` (
`data_month`,`time_range`,`spu`,`reason_level_1`,`reason_level_2`,`reason_issue_category`,
`reject_scene_level_1`,`reject_scene_level_2`,`product_name`,`category_level_1`,`category_level_2`,
`category_level_3`,`article`,`brand_sn`,`brand_name`,`min_price`,`sales_qty`,`reject_qty`,
`reject_amount`,`reject_rate_qty`,`reject_rate_amount`,`source_file`
) VALUES (
%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s
);
"""
def parse_rows(df: pd.DataFrame, data_month: str, source_file: str) -> list[tuple[Any, ...]]:
rows: list[tuple[Any, ...]] = []
for _, r in df.iterrows():
rows.append(
(
data_month,
to_str(r.get("时间")),
to_str(r.get("SPU")),
to_str(r.get("一级拒退原因")),
to_str(r.get("二级拒退原因")),
to_str(r.get("拒退问题分类")),
to_str(r.get("拒退一级场景")),
to_str(r.get("拒退二级场景")),
to_str(r.get("商品名称")),
to_str(r.get("一级品类名称")),
to_str(r.get("二级品类名称")),
to_str(r.get("三级品类名称")),
to_str(r.get("款号")),
to_str(r.get("品牌SN")),
to_str(r.get("品牌名称")),
to_float(r.get("最低到手价")),
to_int(r.get("销售量")),
to_int(r.get("拒退件数")),
to_float(r.get("拒退金额")),
to_float(r.get("拒退率(数量)")),
to_float(r.get("拒退率(金额)")),
source_file,
)
)
return rows
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Load Chinese-header rejection goods xlsx into MySQL.")
p.add_argument("--config", default=r"D:\projects\skills\vip-report\config.yaml")
p.add_argument("--file", required=True)
p.add_argument("--report-year", type=int, default=0)
p.add_argument("--report-month", default="")
return p.parse_args()
def main() -> None:
args = parse_args()
config_path = Path(args.config).resolve()
source_file = Path(args.file).resolve()
cfg = yaml.safe_load(config_path.read_text(encoding="utf-8"))
data_month = resolve_target_month(config_path, args.report_year, args.report_month)
df = pd.read_excel(source_file, sheet_name=0)
required = {"时间", "SPU", "一级拒退原因", "款号", "拒退件数", "拒退率(数量)"}
missing = sorted(required - set(df.columns))
if missing:
raise RuntimeError(f"Missing required columns: {missing}")
time_values = {str(x).strip() for x in df["时间"].dropna().head(20)}
if not any("202603" in x for x in time_values):
raise RuntimeError(f"Unexpected time values for 2026-03 file: {sorted(time_values)[:5]}")
rows = parse_rows(df, data_month, source_file.name)
mysql_cfg = cfg["mysql"]
conn = pymysql.connect(
host=mysql_cfg["host"],
port=int(mysql_cfg.get("port", 3306)),
user=mysql_cfg.get("username") or mysql_cfg.get("user"),
password=mysql_cfg["password"],
database=mysql_cfg["database"],
charset="utf8mb4",
autocommit=False,
)
try:
with conn.cursor() as cur:
cur.execute(create_sql())
cur.execute(f"DELETE FROM `{TABLE_NAME}` WHERE `data_month`=%s", (data_month,))
if rows:
cur.executemany(INSERT_SQL, rows)
cur.execute(f"SELECT COUNT(*) FROM `{TABLE_NAME}` WHERE `data_month`=%s", (data_month,))
month_count = cur.fetchone()[0]
conn.commit()
finally:
conn.close()
print(f"LOAD_OK table={TABLE_NAME} month={data_month} rows={len(rows)} month_count={month_count} file={source_file}")
if __name__ == "__main__":
main()
vip-report/scripts/load_compass_rejection_goods_to_mysql.py 0 → 100644
View file @ 4b9160b1
from __future__ import annotations
import argparse
import re
import sys
from pathlib import Path
from typing import Any
DEPS_DIR = Path(__file__).resolve().parents[2] / ".pydeps"
if DEPS_DIR.exists():
sys.path.insert(0, str(DEPS_DIR))
import pandas as pd
import pymysql
import yaml
TABLE_NAME = "cep_rejection_goods_analysis"
DEFAULT_DOWNLOAD_DIR = r"C:\vipdata"
def parse_month_number(raw: Any) -> int:
if raw is None:
return 1
if isinstance(raw, int):
return min(12, max(1, raw))
text = str(raw).strip()
m = re.search(r"(1[0-2]|0?[1-9])", text)
if m:
return int(m.group(1))
return 1
def resolve_target_month(config_path: Path, report_year_arg: int, report_month_arg: str) -> str:
cfg = yaml.safe_load(config_path.read_text(encoding="utf-8"))
report_cfg = cfg.get("report", {})
year = int(report_year_arg or report_cfg.get("year", 2026))
month = parse_month_number(report_month_arg or report_cfg.get("month_cn", "1"))
return f"{year}{month:02d}"
def month_to_char7(month_key: str) -> str:
return f"{month_key[:4]}-{month_key[4:6]}"
def pick_latest_file(download_dir: Path, month_key: str) -> Path:
files = sorted(download_dir.glob(f"{month_key}_*.xlsx"), key=lambda p: p.stat().st_mtime, reverse=True)
if not files:
raise RuntimeError(f"No xlsx found under {download_dir} for {month_key}")
# Prefer "款号粒度" export for rejection goods analysis.
for f in files:
n = f.name
if "款号粒度" in n and "拒退分析-拒退商品原因" in n:
return f
return files[0]
def to_int(value: Any) -> int | None:
if value is None or (isinstance(value, float) and pd.isna(value)):
return None
s = str(value).strip().replace(",", "")
if not s:
return None
try:
return int(float(s))
except Exception:
return None
def to_float(value: Any) -> float | None:
if value is None or (isinstance(value, float) and pd.isna(value)):
return None
s = str(value).strip().replace(",", "").replace("%", "")
if not s:
return None
try:
return float(s)
except Exception:
return None
def to_str(value: Any) -> str | None:
if value is None or (isinstance(value, float) and pd.isna(value)):
return None
s = str(value).strip()
return s if s else None
def create_sql() -> str:
return f"""
CREATE TABLE IF NOT EXISTS `{TABLE_NAME}` (
`id` BIGINT NOT NULL AUTO_INCREMENT,
`data_month` CHAR(7) NOT NULL,
`time_range` VARCHAR(32) NULL,
`spu` VARCHAR(64) NULL,
`reason_level_1` VARCHAR(128) NULL,
`reason_level_2` VARCHAR(128) NULL,
`reason_issue_category` VARCHAR(128) NULL,
`reject_scene_level_1` VARCHAR(128) NULL,
`reject_scene_level_2` VARCHAR(128) NULL,
`product_name` VARCHAR(512) NULL,
`category_level_1` VARCHAR(128) NULL,
`category_level_2` VARCHAR(128) NULL,
`category_level_3` VARCHAR(128) NULL,
`article` VARCHAR(64) NULL,
`brand_sn` VARCHAR(64) NULL,
`brand_name` VARCHAR(128) NULL,
`min_price` DECIMAL(18,2) NULL,
`sales_qty` INT NULL,
`reject_qty` INT NULL,
`reject_amount` DECIMAL(18,2) NULL,
`reject_rate_qty` DECIMAL(18,6) NULL,
`reject_rate_amount` DECIMAL(18,6) NULL,
`source_file` VARCHAR(255) NOT NULL,
`created_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `idx_data_month` (`data_month`),
KEY `idx_article` (`article`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""
INSERT_SQL = f"""
INSERT INTO `{TABLE_NAME}` (
`data_month`,`time_range`,`spu`,`reason_level_1`,`reason_level_2`,`reason_issue_category`,
`reject_scene_level_1`,`reject_scene_level_2`,`product_name`,`category_level_1`,`category_level_2`,
`category_level_3`,`article`,`brand_sn`,`brand_name`,`min_price`,`sales_qty`,`reject_qty`,
`reject_amount`,`reject_rate_qty`,`reject_rate_amount`,`source_file`
) VALUES (
%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s
);
"""
def parse_rows(df: pd.DataFrame, data_month: str, source_file: str) -> list[tuple[Any, ...]]:
rows: list[tuple[Any, ...]] = []
for _, r in df.iterrows():
rows.append(
(
data_month,
to_str(r.get("时间")),
to_str(r.get("SPU")),
to_str(r.get("一级拒退原因")),
to_str(r.get("二级拒退原因")),
to_str(r.get("拒退问题分类")),
to_str(r.get("拒退一级场景")),
to_str(r.get("拒退二级场景")),
to_str(r.get("商品名称")),
to_str(r.get("一级品类名称")),
to_str(r.get("二级品类名称")),
to_str(r.get("三级品类名称")),
to_str(r.get("款号")),
to_str(r.get("品牌SN")),
to_str(r.get("品牌名称")),
to_float(r.get("最低到手价")),
to_int(r.get("销售量")),
to_int(r.get("拒退件数")),
to_float(r.get("拒退金额")),
to_float(r.get("拒退率(数量)")),
to_float(r.get("拒退率(金额)")),
source_file,
)
)
return rows
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Load rejection goods xlsx into cep_rejection_goods_analysis.")
p.add_argument("--config", default=r"D:\projects\skills\vip-report\config.yaml")
p.add_argument("--download-dir", default=DEFAULT_DOWNLOAD_DIR)
p.add_argument("--report-year", type=int, default=0)
p.add_argument("--report-month", default="")
p.add_argument("--file", default="")
return p.parse_args()
def main() -> None:
args = parse_args()
config_path = Path(args.config).resolve()
cfg = yaml.safe_load(config_path.read_text(encoding="utf-8"))
mysql_cfg = cfg["mysql"]
month_key = resolve_target_month(config_path, args.report_year, args.report_month)
data_month = month_to_char7(month_key)
download_dir = Path(args.download_dir).resolve()
source_file = Path(args.file).resolve() if args.file else pick_latest_file(download_dir, month_key)
if not source_file.exists():
raise RuntimeError(f"source file not found: {source_file}")
df = pd.read_excel(source_file, sheet_name=0)
if df.empty:
raise RuntimeError(f"empty xlsx: {source_file}")
rows = parse_rows(df, data_month, source_file.name)
conn = pymysql.connect(
host=mysql_cfg["host"],
port=int(mysql_cfg.get("port", 3306)),
user=mysql_cfg.get("username") or mysql_cfg.get("user"),
password=mysql_cfg["password"],
database=mysql_cfg["database"],
charset="utf8mb4",
autocommit=False,
)
try:
with conn.cursor() as cur:
cur.execute(create_sql())
cur.execute(f"DELETE FROM `{TABLE_NAME}` WHERE `data_month`=%s", (data_month,))
if rows:
cur.executemany(INSERT_SQL, rows)
cur.execute(f"SELECT COUNT(*) FROM `{TABLE_NAME}` WHERE `data_month`=%s", (data_month,))
month_count = cur.fetchone()[0]
conn.commit()
finally:
conn.close()
print(
f"LOAD_OK table={TABLE_NAME} month={data_month} rows={len(rows)} "
f"month_count={month_count} file={source_file}"
)
if __name__ == "__main__":
main()
vip-report/scripts/load_compass_rejection_scene_to_mysql.py 0 → 100644
View file @ 4b9160b1
from __future__ import annotations
import argparse
import re
import sys
from pathlib import Path
from typing import Any
DEPS_DIR = Path(__file__).resolve().parents[2] / ".pydeps"
if DEPS_DIR.exists():
sys.path.insert(0, str(DEPS_DIR))
import pandas as pd
import pymysql
import yaml
TABLE_NAME = "cep_rejection_scene_analysis"
DEFAULT_DOWNLOAD_DIR = r"C:\vipdata"
def parse_month_number(raw: Any) -> int:
if raw is None:
return 1
if isinstance(raw, int):
return min(12, max(1, raw))
text = str(raw).strip()
m = re.search(r"(1[0-2]|0?[1-9])", text)
if m:
return int(m.group(1))
return 1
def resolve_target_month(config_path: Path, report_year_arg: int, report_month_arg: str) -> str:
cfg = yaml.safe_load(config_path.read_text(encoding="utf-8"))
report_cfg = cfg.get("report", {})
year = int(report_year_arg or report_cfg.get("year", 2026))
month = parse_month_number(report_month_arg or report_cfg.get("month_cn", "1"))
return f"{year}{month:02d}"
def month_to_char7(month_key: str) -> str:
return f"{month_key[:4]}-{month_key[4:6]}"
def pick_latest_file(download_dir: Path, month_key: str) -> Path:
candidates = sorted(download_dir.glob(f"{month_key}_*.xlsx"), key=lambda p: p.stat().st_mtime, reverse=True)
if not candidates:
raise RuntimeError(f"No downloaded xlsx found for month {month_key} under {download_dir}")
return candidates[0]
def normalize_header_name(name: str, idx: int) -> str:
raw = (name or "").strip()
if not raw:
return f"col_{idx:02d}"
k = raw.lower().replace("（", "(").replace("）", ")")
mapping = [
("日期", "biz_date"),
("场景", "scene_name"),
("一级", "category_level_1"),
("二级", "category_level_2"),
("三级", "category_level_3"),
("退货件数", "return_item_count"),
("拒收件数", "reject_item_count"),
("拒退件数", "reject_return_item_count"),
("退货率", "return_rate"),
("拒收率", "reject_rate"),
("拒退率", "reject_return_rate"),
("销售额", "sales_amount"),
("退款金额", "refund_amount"),
("订单数", "order_count"),
("商品件数", "item_count"),
("件数", "item_count"),
("sku", "sku"),
("spu", "spu"),
("品类", "category_name"),
("品牌", "brand_name"),
]
for key, val in mapping:
if key in raw or key in k:
return val if val != "item_count" else f"item_count_{idx:02d}"
ascii_text = re.sub(r"[^a-zA-Z0-9]+", "_", raw).strip("_").lower()
if ascii_text:
return ascii_text
return f"col_{idx:02d}"
def ensure_unique(names: list[str]) -> list[str]:
used: dict[str, int] = {}
out: list[str] = []
for n in names:
base = n
if base not in used:
used[base] = 1
out.append(base)
continue
used[base] += 1
out.append(f"{base}_{used[base]}")
return out
def infer_sql_type(values: pd.Series, col_name: str) -> str:
if col_name == "biz_date":
return "DATE"
sample = values.dropna().astype(str).str.strip()
if sample.empty:
return "VARCHAR(255)"
if sample.str.match(r"^\d{4}-\d{2}-\d{2}$").mean() > 0.8:
return "DATE"
if sample.str.match(r"^-?\d+$").mean() > 0.8:
return "BIGINT"
if sample.str.match(r"^-?\d+(\.\d+)?%?$").mean() > 0.8:
return "DECIMAL(18,6)"
max_len = int(sample.str.len().max())
if max_len <= 255:
return "VARCHAR(255)"
if max_len <= 1024:
return "VARCHAR(1024)"
return "TEXT"
def clean_value(v: Any, sql_type: str) -> Any:
if pd.isna(v):
return None
s = str(v).strip()
if not s:
return None
if sql_type == "DATE":
m = re.match(r"^(\d{4})[-/](\d{1,2})[-/](\d{1,2})$", s)
if m:
return f"{m.group(1)}-{int(m.group(2)):02d}-{int(m.group(3)):02d}"
m2 = re.match(r"^(\d{4})(\d{2})(\d{2})$", re.sub(r"\D", "", s))
if m2:
return f"{m2.group(1)}-{m2.group(2)}-{m2.group(3)}"
return None
if sql_type.startswith("DECIMAL"):
s2 = s.replace(",", "").replace("%", "")
try:
return float(s2)
except Exception:
return None
if sql_type == "BIGINT":
s2 = s.replace(",", "")
try:
return int(float(s2))
except Exception:
return None
return s
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Create cep rejection-scene table and load latest downloaded xlsx.")
p.add_argument("--config", default=r"D:\projects\skills\vip-report\config.yaml")
p.add_argument("--download-dir", default=DEFAULT_DOWNLOAD_DIR)
p.add_argument("--report-year", type=int, default=0)
p.add_argument("--report-month", default="")
p.add_argument("--file", default="")
return p.parse_args()
def run_loader(config_path: Path, download_dir: Path, report_year: int, report_month: str, file_arg: str) -> None:
cfg = yaml.safe_load(config_path.read_text(encoding="utf-8"))
mysql_cfg = cfg.get("mysql", {})
conn_cfg = {
"host": mysql_cfg.get("host"),
"port": int(mysql_cfg.get("port", 3306)),
"user": mysql_cfg.get("username") or mysql_cfg.get("user"),
"password": mysql_cfg.get("password"),
"database": mysql_cfg.get("database"),
"charset": "utf8mb4",
"autocommit": False,
}
month_key = resolve_target_month(config_path, report_year, report_month)
source_file = Path(file_arg) if file_arg else pick_latest_file(download_dir, month_key)
if not source_file.exists():
raise RuntimeError(f"source file not found: {source_file}")
df = pd.read_excel(source_file, sheet_name=0)
if df.empty:
raise RuntimeError(f"source file has no rows: {source_file}")
raw_headers = [str(c).strip() for c in df.columns]
english_headers = ensure_unique([normalize_header_name(h, i + 1) for i, h in enumerate(raw_headers)])
df.columns = english_headers
if "biz_date" not in df.columns:
df.insert(0, "biz_date", None)
df.insert(1, "data_month", month_to_char7(month_key))
df.insert(2, "source_file", source_file.name)
sql_types: dict[str, str] = {}
for c in df.columns:
if c in {"data_month", "source_file"}:
sql_types[c] = "VARCHAR(64)" if c == "data_month" else "VARCHAR(255)"
else:
sql_types[c] = infer_sql_type(df[c], c)
cols_sql = ",\n ".join([f"`{c}` {sql_types[c]} NULL" for c in df.columns])
create_sql = f"""
CREATE TABLE IF NOT EXISTS `{TABLE_NAME}` (
`id` BIGINT NOT NULL AUTO_INCREMENT,
{cols_sql},
`created_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `idx_data_month` (`data_month`),
KEY `idx_biz_date` (`biz_date`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""
insert_cols = [f"`{c}`" for c in df.columns]
placeholders = ",".join(["%s"] * len(df.columns))
insert_sql = f"INSERT INTO `{TABLE_NAME}` ({','.join(insert_cols)}) VALUES ({placeholders})"
rows: list[tuple[Any, ...]] = []
for _, r in df.iterrows():
row_vals: list[Any] = []
for c in df.columns:
row_vals.append(clean_value(r[c], sql_types[c]))
rows.append(tuple(row_vals))
conn = pymysql.connect(**conn_cfg)
try:
with conn.cursor() as cur:
cur.execute(create_sql)
cur.execute(f"SHOW COLUMNS FROM `{TABLE_NAME}`")
existing_cols = {x[0] for x in cur.fetchall()}
for c in df.columns:
if c in existing_cols:
continue
cur.execute(f"ALTER TABLE `{TABLE_NAME}` ADD COLUMN `{c}` {sql_types[c]} NULL")
cur.execute(f"DELETE FROM `{TABLE_NAME}` WHERE `data_month`=%s", (month_to_char7(month_key),))
if rows:
cur.executemany(insert_sql, rows)
cur.execute(f"SELECT COUNT(*) FROM `{TABLE_NAME}` WHERE `data_month`=%s", (month_to_char7(month_key),))
month_count = cur.fetchone()[0]
conn.commit()
finally:
conn.close()
print(
f"LOAD_OK table={TABLE_NAME} month={month_to_char7(month_key)} rows={len(rows)} "
f"month_count={month_count} file={source_file}"
)
print("COLUMN_MAP:")
for zh, en in zip(raw_headers, english_headers):
print(f" {zh} -> {en}")
def main() -> None:
args = parse_args()
run_loader(
config_path=Path(args.config).resolve(),
download_dir=Path(args.download_dir).resolve(),
report_year=args.report_year,
report_month=args.report_month,
file_arg=args.file,
)
if __name__ == "__main__":
main()
vip-report/scripts/load_traffic_downloads_to_mysql.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/load_vis_live_session_to_mysql.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/load_vis_vendor_goods_effect_to_mysql.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/render_template_com.ps1
View file @ 4b9160b1
param( param(
[string]$TemplatePath = "C:\Users\niuniu\Desktop\Report.pptx", [string]$TemplatePath = "C:\Users\niuniu\Desktop\Report.pptx",
[string]$OutputPath = "C:\workspace\cursor\output\vip-report\generated-from-template.pptx", [string]$OutputPath = "C:\workspace\cursor\output\vip-report\generated-from-template.pptx",
[string]$OperationsPath = "" [string]$OperationsPath = ""
...@@ -6,12 +6,12 @@ param( ...@@ -6,12 +6,12 @@ param(
$ErrorActionPreference = "Stop" $ErrorActionPreference = "Stop"
# 关键行注释：这里是关键处理节点。 # 鍏抽敭琛屾敞閲婏細杩欓噷鏄叧閿鐞嗚妭鐐广€?
# 函数说明：封装Get-ShapeChildren方法的核心处理流程。 # 鍑芥暟璇存槑锛氬皝瑁匞et-ShapeChildren鏂规硶鐨勬牳蹇冨鐞嗘祦绋嬨€?
function Get-ShapeChildren { function Get-ShapeChildren {
param([Parameter(Mandatory = $true)]$Shape) param([Parameter(Mandatory = $true)]$Shape)
$children = @() $children = @()
# 关键行注释：进入核心执行区，并在 finally 中释放 PowerPoint 资源。 # 鍏抽敭琛屾敞閲婏細杩涘叆鏍稿績鎵ц鍖猴紝骞跺湪 finally 涓噴鏀?PowerPoint 璧勬簮銆?
try { try {
$count = $Shape.GroupItems.Count $count = $Shape.GroupItems.Count
for ($i = 1; $i -le $count; $i++) { for ($i = 1; $i -le $count; $i++) {
...@@ -24,8 +24,8 @@ function Get-ShapeChildren { ...@@ -24,8 +24,8 @@ function Get-ShapeChildren {
return $children return $children
} }
# 关键行注释：这里是关键处理节点。 # 鍏抽敭琛屾敞閲婏細杩欓噷鏄叧閿鐞嗚妭鐐广€?
# 函数说明：封装Find-Shape方法的核心处理流程。 # 鍑芥暟璇存槑锛氬皝瑁匜ind-Shape鏂规硶鐨勬牳蹇冨鐞嗘祦绋嬨€?
function Find-Shape { function Find-Shape {
param( param(
[Parameter(Mandatory = $true)]$Slide, [Parameter(Mandatory = $true)]$Slide,
...@@ -35,6 +35,14 @@ function Find-Shape { ...@@ -35,6 +35,14 @@ function Find-Shape {
[string]$ContainsText [string]$ContainsText
) )
$targetNumber = $null
if ($ShapeName) {
$numMatch = [regex]::Match($ShapeName, "(\d+)\s*$")
if ($numMatch.Success) {
$targetNumber = $numMatch.Groups[1].Value
}
}
$candidates = @() $candidates = @()
foreach ($shape in $Slide.Shapes) { foreach ($shape in $Slide.Shapes) {
$candidates += $shape $candidates += $shape
...@@ -46,12 +54,23 @@ function Find-Shape { ...@@ -46,12 +54,23 @@ function Find-Shape {
return $shape return $shape
} }
if ($ShapeName -and $shape.Name -eq $ShapeName) { if ($ShapeName) {
return $shape if ($shape.Name -eq $ShapeName) {
return $shape
}
if ($targetNumber) {
$shapeNumMatch = [regex]::Match([string]$shape.Name, "(\d+)\s*$")
if ($shapeNumMatch.Success -and $shapeNumMatch.Groups[1].Value -eq $targetNumber) {
$shapeNameLower = ([string]$shape.Name).ToLowerInvariant()
if ($shapeNameLower.Contains("picture")) {
return $shape
}
}
}
} }
$text = $null $text = $null
# 关键行注释：进入核心执行区，并在 finally 中释放 PowerPoint 资源。 # 鍏抽敭琛屾敞閲婏細杩涘叆鏍稿績鎵ц鍖猴紝骞跺湪 finally 涓噴鏀?PowerPoint 璧勬簮銆?
try { try {
if ($shape.HasTextFrame -eq -1 -and $shape.TextFrame.HasText -eq -1) { if ($shape.HasTextFrame -eq -1 -and $shape.TextFrame.HasText -eq -1) {
$text = $shape.TextFrame.TextRange.Text $text = $shape.TextFrame.TextRange.Text
...@@ -72,8 +91,8 @@ function Find-Shape { ...@@ -72,8 +91,8 @@ function Find-Shape {
return $null return $null
} }
# 关键行注释：这里是关键处理节点。 # 鍏抽敭琛屾敞閲婏細杩欓噷鏄叧閿鐞嗚妭鐐广€?
# 函数说明：封装Set-ShapeText方法的核心处理流程。 # 鍑芥暟璇存槑锛氬皝瑁匰et-ShapeText鏂规硶鐨勬牳蹇冨鐞嗘祦绋嬨€?
function Set-ShapeText { function Set-ShapeText {
param( param(
[Parameter(Mandatory = $true)]$Shape, [Parameter(Mandatory = $true)]$Shape,
...@@ -87,8 +106,8 @@ function Set-ShapeText { ...@@ -87,8 +106,8 @@ function Set-ShapeText {
$Shape.TextFrame.TextRange.Text = $NewText $Shape.TextFrame.TextRange.Text = $NewText
} }
# 关键行注释：这里是关键处理节点。 # 鍏抽敭琛屾敞閲婏細杩欓噷鏄叧閿鐞嗚妭鐐广€?
# 函数说明：封装Replace-PictureShape方法的核心处理流程。 # 鍑芥暟璇存槑锛氬皝瑁匯eplace-PictureShape鏂规硶鐨勬牳蹇冨鐞嗘祦绋嬨€?
function Replace-PictureShape { function Replace-PictureShape {
param( param(
[Parameter(Mandatory = $true)]$Slide, [Parameter(Mandatory = $true)]$Slide,
...@@ -112,7 +131,7 @@ function Replace-PictureShape { ...@@ -112,7 +131,7 @@ function Replace-PictureShape {
while ($newShape.ZOrderPosition -lt $z) { while ($newShape.ZOrderPosition -lt $z) {
$newShape.ZOrder(1) | Out-Null $newShape.ZOrder(1) | Out-Null
} }
# 关键行注释：进入核心执行区，并在 finally 中释放 PowerPoint 资源。 # 鍏抽敭琛屾敞閲婏細杩涘叆鏍稿績鎵ц鍖猴紝骞跺湪 finally 涓噴鏀?PowerPoint 璧勬簮銆?
try { try {
$newShape.Name = $name $newShape.Name = $name
} catch { } catch {
...@@ -123,6 +142,10 @@ if (-not (Test-Path -LiteralPath $TemplatePath)) { ...@@ -123,6 +142,10 @@ if (-not (Test-Path -LiteralPath $TemplatePath)) {
throw "TemplatePath not found: $TemplatePath" throw "TemplatePath not found: $TemplatePath"
} }
$templateResolved = [System.IO.Path]::GetFullPath($TemplatePath)
$outputResolved = [System.IO.Path]::GetFullPath($OutputPath)
$isInPlaceUpdate = $templateResolved.Equals($outputResolved, [System.StringComparison]::OrdinalIgnoreCase)
$ops = [pscustomobject]@{ $ops = [pscustomobject]@{
replace_text = @() replace_text = @()
replace_tables = @() replace_tables = @()
...@@ -159,10 +182,12 @@ if ($outDir -and -not (Test-Path -LiteralPath $outDir)) { ...@@ -159,10 +182,12 @@ if ($outDir -and -not (Test-Path -LiteralPath $outDir)) {
New-Item -ItemType Directory -Path $outDir -Force | Out-Null New-Item -ItemType Directory -Path $outDir -Force | Out-Null
} }
Copy-Item -LiteralPath $TemplatePath -Destination $OutputPath -Force if (-not $isInPlaceUpdate) {
Copy-Item -LiteralPath $TemplatePath -Destination $OutputPath -Force
}
if ($replaceText.Count -eq 0 -and $replaceTables.Count -eq 0 -and $replaceCharts.Count -eq 0 -and $replaceImages.Count -eq 0) { if ($replaceText.Count -eq 0 -and $replaceTables.Count -eq 0 -and $replaceCharts.Count -eq 0 -and $replaceImages.Count -eq 0) {
# 无替换操作时直接复制模板，避免 PowerPoint 重新保存造成包结构差异。 # 鏃犳浛鎹㈡搷浣滄椂鐩存帴澶嶅埗妯℃澘锛岄伩鍏?PowerPoint 閲嶆柊淇濆瓨閫犳垚鍖呯粨鏋勫樊寮傘€?
Write-Output $OutputPath Write-Output $OutputPath
return return
} }
...@@ -170,7 +195,7 @@ if ($replaceText.Count -eq 0 -and $replaceTables.Count -eq 0 -and $replaceCharts ...@@ -170,7 +195,7 @@ if ($replaceText.Count -eq 0 -and $replaceTables.Count -eq 0 -and $replaceCharts
$ppt = $null $ppt = $null
$pres = $null $pres = $null
# 关键行注释：进入核心执行区，并在 finally 中释放 PowerPoint 资源。 # 鍏抽敭琛屾敞閲婏細杩涘叆鏍稿績鎵ц鍖猴紝骞跺湪 finally 涓噴鏀?PowerPoint 璧勬簮銆?
try { try {
$ppt = New-Object -ComObject PowerPoint.Application $ppt = New-Object -ComObject PowerPoint.Application
$pres = $ppt.Presentations.Open($OutputPath, 0, 0, 0) $pres = $ppt.Presentations.Open($OutputPath, 0, 0, 0)
...@@ -249,3 +274,4 @@ try { ...@@ -249,3 +274,4 @@ try {
try { $ppt.Quit() | Out-Null } catch {} try { $ppt.Quit() | Out-Null } catch {}
} }
} }
vip-report/scripts/run_flow12_traffic_pipeline.py 0 → 100644
View file @ 4b9160b1
from __future__ import annotations
import argparse
import subprocess
import sys
from pathlib import Path
SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPT_DIR))
from runtime_config import load_runtime_config
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Run flow12 traffic pipeline in one command.")
p.add_argument("--python", default=sys.executable, help="Python executable path")
p.add_argument("--config-source", choices=["yaml", "db"], default="db")
p.add_argument("--config", default="", help="yaml path when config-source=yaml")
p.add_argument("--config-table", default="cepconfiguration")
p.add_argument("--config-key", default="vip-report")
p.add_argument("--mysql-host", default="192.168.138.182")
p.add_argument("--mysql-port", type=int, default=3306)
p.add_argument("--mysql-user", default="ceptest")
p.add_argument("--mysql-password", default="ck@123456")
p.add_argument("--mysql-database", default="ckc_cep_db_test")
p.add_argument("--download-dir", default=r"C:\vipdata")
return p.parse_args()
def main() -> None:
args = parse_args()
root = Path(__file__).resolve().parent
common = [
"--config-source",
args.config_source,
"--config-table",
args.config_table,
"--config-key",
args.config_key,
"--mysql-host",
args.mysql_host,
"--mysql-port",
str(args.mysql_port),
"--mysql-user",
args.mysql_user,
"--mysql-password",
args.mysql_password,
"--mysql-database",
args.mysql_database,
]
if args.config_source == "yaml":
config_path = args.config or str((root.parent / "config.yaml").resolve())
common = ["--config", config_path]
runtime_cfg = load_runtime_config(args, default_config=Path(args.config) if args.config else (root.parent / "config.yaml"))
workdir = Path(runtime_cfg["paths"]["workdir"]).resolve()
steps: list[list[str]] = [
[
args.python,
str(root / "sync_compass_traffic_source_download.py"),
*common,
"--download-dir",
args.download_dir,
],
[args.python, str(root / "load_traffic_downloads_to_mysql.py"), *common],
[args.python, str(root / "sync_cep_traffic_assets.py"), *common],
[args.python, str(root / "sync_cep_site_traffic_entry_assets.py"), *common],
[args.python, str(root / "sync_cep_traffic_summary_assets.py"), *common],
[
args.python,
str(root / "compose_uv_lfl_report.py"),
"--table-image",
str(workdir / "assets" / "cep-traffic" / "cep-traffic-summary.png"),
"--line-image",
str(workdir / "assets" / "cep-site-traffic-entry" / "cep-site-traffic-entry-lfl.png"),
"--entry-image",
str(workdir / "assets" / "cep-traffic-summary" / "cep-traffic-summary-lfl.png"),
"--output",
str(workdir / "assets" / "uv-lfl" / "uv-lfl-report.png"),
],
]
for i, cmd in enumerate(steps, start=1):
print(f"[STEP {i}/{len(steps)}] {' '.join(cmd)}")
result = subprocess.run(cmd, text=True, capture_output=True)
if result.stdout:
print(result.stdout)
if result.returncode != 0:
err = (result.stderr or "").strip()
out = (result.stdout or "").strip()
# Keep message bounded but include enough detail for ShadowBot surfaced errors.
err_tail = err[-4000:] if len(err) > 4000 else err
out_tail = out[-2000:] if len(out) > 2000 else out
raise RuntimeError(
"STEP_FAILED\n"
f"step={i}\n"
f"command={' '.join(cmd)}\n"
f"returncode={result.returncode}\n"
f"stdout_tail=\n{out_tail}\n"
f"stderr_tail=\n{err_tail}\n"
)
final_image = workdir / "assets" / "uv-lfl" / "uv-lfl-report.png"
print(f"FINAL_IMAGE={final_image}")
print("FLOW12_PIPELINE_OK")
if __name__ == "__main__":
main()
vip-report/scripts/run_flow14_livestream_pipeline.py 0 → 100644
View file @ 4b9160b1
from __future__ import annotations
import argparse
import subprocess
import sys
from pathlib import Path
SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPT_DIR))
from runtime_config import load_runtime_config
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Run flow14 livestream pipeline in one command.")
p.add_argument("--python", default=sys.executable, help="Python executable path")
p.add_argument("--config-source", choices=["yaml", "db"], default="db")
p.add_argument("--config", default="", help="yaml path when config-source=yaml")
p.add_argument("--config-table", default="cepconfiguration")
p.add_argument("--config-key", default="vip-report")
p.add_argument("--mysql-host", default="192.168.138.182")
p.add_argument("--mysql-port", type=int, default=3306)
p.add_argument("--mysql-user", default="ceptest")
p.add_argument("--mysql-password", default="ck@123456")
p.add_argument("--mysql-database", default="ckc_cep_db_test")
p.add_argument("--download-dir", default=r"C:\vipdata")
p.add_argument("--skip-compose", action="store_true", help="Only download/import/render 2 images, skip final compose")
return p.parse_args()
def build_common_args(args: argparse.Namespace) -> list[str]:
common = [
"--config-source",
args.config_source,
"--config-table",
args.config_table,
"--config-key",
args.config_key,
"--mysql-host",
args.mysql_host,
"--mysql-port",
str(args.mysql_port),
"--mysql-user",
args.mysql_user,
"--mysql-password",
args.mysql_password,
"--mysql-database",
args.mysql_database,
]
if args.config_source == "yaml":
config_path = args.config or str((SCRIPT_DIR.parent / "config.yaml").resolve())
return ["--config", config_path]
return common
def run_step(step_index: int, total: int, cmd: list[str]) -> None:
print(f"[STEP {step_index}/{total}] {' '.join(cmd)}")
result = subprocess.run(cmd, text=True, capture_output=True)
if result.stdout:
print(result.stdout)
if result.returncode != 0:
err = (result.stderr or "").strip()
out = (result.stdout or "").strip()
err_tail = err[-4000:] if len(err) > 4000 else err
out_tail = out[-2000:] if len(out) > 2000 else out
raise RuntimeError(
"STEP_FAILED\n"
f"step={step_index}\n"
f"command={' '.join(cmd)}\n"
f"returncode={result.returncode}\n"
f"stdout_tail=\n{out_tail}\n"
f"stderr_tail=\n{err_tail}\n"
)
def main() -> None:
args = parse_args()
root = SCRIPT_DIR
common = build_common_args(args)
runtime_cfg = load_runtime_config(
args,
default_config=Path(args.config) if args.config else (root.parent / "config.yaml"),
)
workdir = Path(runtime_cfg["paths"]["workdir"]).resolve()
steps: list[list[str]] = [
[
args.python,
str(root / "sync_vis_live_session_downloads.py"),
*common,
"--download-dir",
args.download_dir,
],
[args.python, str(root / "load_vis_live_session_to_mysql.py"), *common],
[args.python, str(root / "sync_cep_live_session_assets.py"), *common],
[args.python, str(root / "sync_cep_live_session_trend_assets.py"), *common],
]
if not args.skip_compose:
steps.append(
[
args.python,
str(root / "compose_live_session_report.py"),
*common,
"--summary-image",
str(workdir / "assets" / "cep-live-session" / "cep-live-session-summary.png"),
"--trend-image",
str(workdir / "assets" / "cep-live-session" / "cep-live-session-sales-penetration-trend.png"),
"--pivot-image",
str(workdir / "assets" / "cep-live-goods-effect" / "cep-live-goods-effect-pivot.png"),
"--output",
str(workdir / "assets" / "cep-live-session" / "cep-live-session-combined.png"),
]
)
total = len(steps)
for i, cmd in enumerate(steps, start=1):
run_step(i, total, cmd)
print(f"FINAL_IMAGE={workdir / 'assets' / 'cep-live-session' / 'cep-live-session-combined.png'}")
print("FLOW14_PIPELINE_OK")
if __name__ == "__main__":
main()
vip-report/scripts/run_flow15_live_goods_effect_pipeline.py 0 → 100644
View file @ 4b9160b1
from __future__ import annotations
import argparse
import subprocess
import sys
from pathlib import Path
SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPT_DIR))
from runtime_config import load_runtime_config
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Run flow15 live-goods-effect pipeline in one command.")
p.add_argument("--python", default=sys.executable, help="Python executable path")
p.add_argument("--config-source", choices=["yaml", "db"], default="db")
p.add_argument("--config", default="", help="yaml path when config-source=yaml")
p.add_argument("--config-table", default="cepconfiguration")
p.add_argument("--config-key", default="vip-report")
p.add_argument("--mysql-host", default="192.168.138.182")
p.add_argument("--mysql-port", type=int, default=3306)
p.add_argument("--mysql-user", default="ceptest")
p.add_argument("--mysql-password", default="ck@123456")
p.add_argument("--mysql-database", default="ckc_cep_db_test")
p.add_argument("--download-dir", default=r"C:\vipdata")
p.add_argument("--skip-compose", action="store_true")
return p.parse_args()
def build_common_args(args: argparse.Namespace) -> list[str]:
if args.config_source == "yaml":
config_path = args.config or str((SCRIPT_DIR.parent / "config.yaml").resolve())
return ["--config", config_path]
return [
"--config-source",
args.config_source,
"--config-table",
args.config_table,
"--config-key",
args.config_key,
"--mysql-host",
args.mysql_host,
"--mysql-port",
str(args.mysql_port),
"--mysql-user",
args.mysql_user,
"--mysql-password",
args.mysql_password,
"--mysql-database",
args.mysql_database,
]
def run_step(step_index: int, total: int, cmd: list[str]) -> None:
print(f"[STEP {step_index}/{total}] {' '.join(cmd)}")
result = subprocess.run(cmd, text=True, capture_output=True)
if result.stdout:
print(result.stdout)
if result.returncode != 0:
err = (result.stderr or "").strip()
out = (result.stdout or "").strip()
err_tail = err[-4000:] if len(err) > 4000 else err
out_tail = out[-2000:] if len(out) > 2000 else out
raise RuntimeError(
"STEP_FAILED\n"
f"step={step_index}\n"
f"command={' '.join(cmd)}\n"
f"returncode={result.returncode}\n"
f"stdout_tail=\n{out_tail}\n"
f"stderr_tail=\n{err_tail}\n"
)
def main() -> None:
args = parse_args()
root = SCRIPT_DIR
common = build_common_args(args)
runtime_cfg = load_runtime_config(
args,
default_config=Path(args.config) if args.config else (root.parent / "config.yaml"),
)
workdir = Path(runtime_cfg["paths"]["workdir"]).resolve()
steps: list[list[str]] = [
[
args.python,
str(root / "sync_vis_vendor_goods_effect_download.py"),
*common,
"--download-dir",
args.download_dir,
"--no-auto-load",
],
[
args.python,
str(root / "load_vis_vendor_goods_effect_to_mysql.py"),
*common,
"--download-dir",
args.download_dir,
],
[args.python, str(root / "sync_cep_live_goods_effect_summary_assets.py"), *common],
[args.python, str(root / "sync_cep_live_goods_effect_pivot_assets.py"), *common],
[args.python, str(root / "sync_cep_live_goods_effect_top5_assets.py"), *common],
]
if not args.skip_compose:
steps.append(
[
args.python,
str(root / "compose_live_goods_effect_report.py"),
*common,
"--summary-image",
str(workdir / "assets" / "cep-live-goods-effect" / "cep-live-goods-effect-summary.png"),
"--top5-image",
str(workdir / "assets" / "cep-live-goods-effect" / "cep-live-goods-effect-top5.png"),
"--output",
str(workdir / "assets" / "cep-live-goods-effect" / "cep-live-goods-effect-combined.png"),
]
)
total = len(steps)
for i, cmd in enumerate(steps, start=1):
run_step(i, total, cmd)
print(f"FINAL_IMAGE={workdir / 'assets' / 'cep-live-goods-effect' / 'cep-live-goods-effect-combined.png'}")
print("FLOW15_PIPELINE_OK")
if __name__ == "__main__":
main()
vip-report/scripts/run_flow17_membership_pipeline.py 0 → 100644
View file @ 4b9160b1
from __future__ import annotations
import argparse
import subprocess
import sys
from pathlib import Path
SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPT_DIR))
from runtime_config import load_runtime_config
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Run flow17 membership pipeline in one command.")
p.add_argument("--python", default=sys.executable, help="Python executable path")
p.add_argument("--config-source", choices=["yaml", "db"], default="db")
p.add_argument("--config", default="", help="yaml path when config-source=yaml")
p.add_argument("--config-table", default="cepconfiguration")
p.add_argument("--config-key", default="vip-report")
p.add_argument("--mysql-host", default="192.168.138.182")
p.add_argument("--mysql-port", type=int, default=3306)
p.add_argument("--mysql-user", default="ceptest")
p.add_argument("--mysql-password", default="ck@123456")
p.add_argument("--mysql-database", default="ckc_cep_db_test")
p.add_argument("--session", default="vip-report-monthly-sales")
p.add_argument("--skip-compose", action="store_true")
return p.parse_args()
def build_common_args(args: argparse.Namespace) -> list[str]:
if args.config_source == "yaml":
config_path = args.config or str((SCRIPT_DIR.parent / "config.yaml").resolve())
return ["--config", config_path]
return [
"--config-source",
args.config_source,
"--config-table",
args.config_table,
"--config-key",
args.config_key,
"--mysql-host",
args.mysql_host,
"--mysql-port",
str(args.mysql_port),
"--mysql-user",
args.mysql_user,
"--mysql-password",
args.mysql_password,
"--mysql-database",
args.mysql_database,
]
def run_step(step_index: int, total: int, cmd: list[str]) -> None:
print(f"[STEP {step_index}/{total}] {' '.join(cmd)}")
result = subprocess.run(cmd, text=True, capture_output=True)
if result.stdout:
print(result.stdout)
if result.returncode != 0:
err = (result.stderr or "").strip()
out = (result.stdout or "").strip()
err_tail = err[-4000:] if len(err) > 4000 else err
out_tail = out[-2000:] if len(out) > 2000 else out
raise RuntimeError(
"STEP_FAILED\n"
f"step={step_index}\n"
f"command={' '.join(cmd)}\n"
f"returncode={result.returncode}\n"
f"stdout_tail=\n{out_tail}\n"
f"stderr_tail=\n{err_tail}\n"
)
def main() -> None:
args = parse_args()
root = SCRIPT_DIR
common = build_common_args(args)
runtime_cfg = load_runtime_config(
args,
default_config=Path(args.config) if args.config else (root.parent / "config.yaml"),
)
workdir = Path(runtime_cfg["paths"]["workdir"]).resolve()
report_cfg = runtime_cfg.get("report", {})
year = int(report_cfg.get("year", 2026))
month_raw = str(report_cfg.get("month_cn", "1"))
month = "".join([c for c in month_raw if c.isdigit()]) or "1"
month = month.zfill(2)
ym = f"{year}{month}"
steps: list[list[str]] = [
[args.python, str(root / "sync_vis_member_overview_capture.py"), *common, "--session", args.session],
[args.python, str(root / "sync_vis_member_consumption_capture.py"), *common, "--session", args.session],
[args.python, str(root / "sync_vis_member_order_distribution_capture.py"), *common, "--session", args.session],
[args.python, str(root / "sync_vis_member_channel_trend_capture.py"), *common, "--session", args.session],
]
if not args.skip_compose:
steps.append([args.python, str(root / "compose_member_overview_report.py"), *common])
total = len(steps)
for i, cmd in enumerate(steps, start=1):
run_step(i, total, cmd)
print(f"FINAL_IMAGE={workdir / 'assets' / 'cep-member-overview' / ('member-overview-combined-' + ym + '.png')}")
print("FLOW17_PIPELINE_OK")
if __name__ == "__main__":
main()
vip-report/scripts/run_flow18_rejection_pipeline.py 0 → 100644
View file @ 4b9160b1
from __future__ import annotations
import argparse
import json
import sys
import time
from pathlib import Path
import yaml
SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPT_DIR))
from runtime_config import add_config_source_args, load_runtime_config
def parse_month_number(raw: object) -> int:
if raw is None:
return 1
text = str(raw).strip()
digits = "".join(ch for ch in text if ch.isdigit())
if digits:
m = int(digits[-2:] if len(digits) > 2 else digits)
return max(1, min(12, m))
return 1
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Run flow18 rejection pipeline in one command.")
p.add_argument("--python", default=sys.executable, help="Python executable path")
add_config_source_args(p, default_config=Path(r"D:\projects\skills\vip-report\config.yaml"))
p.add_argument("--session", default="vip-report-monthly-sales")
p.add_argument("--download-dir", default=r"C:\vipdata")
p.add_argument("--report-year", type=int, default=0)
p.add_argument("--report-month", default="")
p.add_argument("--max-retries", type=int, default=15)
p.add_argument("--scene-file", default="")
p.add_argument("--goods-file", default="")
p.add_argument("--skip-compose", action="store_true")
return p.parse_args()
def run_step(step_index: int, total: int, cmd: list[str]) -> tuple[str, dict[str, object] | None]:
import subprocess
print(f"[STEP {step_index}/{total}] {' '.join(cmd)}")
result = subprocess.run(cmd, text=True, capture_output=True)
stdout_text = result.stdout or ""
if stdout_text.strip():
print(stdout_text)
if result.returncode != 0:
err = (result.stderr or "").strip()
out = stdout_text.strip()
err_tail = err[-4000:] if len(err) > 4000 else err
out_tail = out[-2000:] if len(out) > 2000 else out
raise RuntimeError(
"STEP_FAILED\n"
f"step={step_index}\n"
f"command={' '.join(cmd)}\n"
f"returncode={result.returncode}\n"
f"stdout_tail=\n{out_tail}\n"
f"stderr_tail=\n{err_tail}\n"
)
parsed_json: dict[str, object] | None = None
for line in stdout_text.splitlines():
s = line.strip()
if s.startswith("{") and s.endswith("}"):
try:
parsed_json = json.loads(s)
except Exception:
pass
return stdout_text, parsed_json
def build_month_values(cfg: dict[str, object], report_year_arg: int, report_month_arg: str) -> tuple[int, str]:
report_cfg = cfg.get("report", {}) if isinstance(cfg.get("report"), dict) else {}
year = int(report_year_arg or report_cfg.get("year", 2026))
month_num = parse_month_number(report_month_arg or report_cfg.get("month_cn", "1"))
return year, f"{month_num:02d}"
def main() -> None:
args = parse_args()
default_config_path = Path(args.config or r"D:\projects\skills\vip-report\config.yaml").resolve()
cfg = load_runtime_config(args, default_config=default_config_path)
workdir = Path(cfg["paths"]["workdir"]).resolve()
year, month = build_month_values(cfg, args.report_year, args.report_month)
tmp_config_path: Path | None = None
common_cfg_args: list[str]
if args.config_source == "yaml":
common_cfg_args = ["--config", str(default_config_path)]
else:
tmp_config_path = workdir / f"tmp-flow18-config-{year}{month}-{time.time_ns()}.yaml"
tmp_config_path.write_text(yaml.safe_dump(cfg, allow_unicode=True, sort_keys=False), encoding="utf-8")
common_cfg_args = ["--config", str(tmp_config_path)]
download_dir = Path(args.download_dir).resolve()
download_dir.mkdir(parents=True, exist_ok=True)
try:
steps: list[list[str]] = [
[
args.python,
str(SCRIPT_DIR / "sync_compass_rejection_analysis_capture.py"),
*common_cfg_args,
"--session",
args.session,
"--report-year",
str(year),
"--report-month",
month,
],
[
args.python,
str(SCRIPT_DIR / "sync_compass_rejection_scene_download.py"),
*common_cfg_args,
"--session",
args.session,
"--report-year",
str(year),
"--report-month",
month,
"--download-dir",
str(download_dir),
"--max-retries",
str(int(args.max_retries)),
],
]
total_steps = 7 if args.skip_compose else 8
_, step2_json = run_step(1, total_steps, steps[0])
_, step_scene_json = run_step(2, total_steps, steps[1])
scene_file = str(args.scene_file or ((step_scene_json or {}).get("savedAs") or ""))
step3_cmd = [
args.python,
str(SCRIPT_DIR / "load_compass_rejection_scene_to_mysql.py"),
*common_cfg_args,
"--download-dir",
str(download_dir),
"--report-year",
str(year),
"--report-month",
month,
]
if scene_file:
step3_cmd.extend(["--file", scene_file])
run_step(3, total_steps, step3_cmd)
run_step(
4,
total_steps,
[
args.python,
str(SCRIPT_DIR / "sync_cep_rejection_lfl_assets.py"),
*common_cfg_args,
"--report-year",
str(year),
"--report-month",
month,
],
)
_, step_goods_json = run_step(
5,
total_steps,
[
args.python,
str(SCRIPT_DIR / "sync_compass_rejection_goods_download.py"),
*common_cfg_args,
"--session",
args.session,
"--report-year",
str(year),
"--report-month",
month,
"--download-dir",
str(download_dir),
"--max-retries",
str(int(args.max_retries)),
],
)
goods_file = str(args.goods_file or ((step_goods_json or {}).get("savedAs") or ""))
if not goods_file:
raise RuntimeError("STEP_FAILED\nstep=6\ngoods download output missing savedAs path")
run_step(
6,
total_steps,
[
args.python,
str(SCRIPT_DIR / "load_compass_rejection_goods_cn_to_mysql.py"),
*common_cfg_args,
"--file",
goods_file,
"--report-year",
str(year),
"--report-month",
month,
],
)
run_step(
7,
total_steps,
[
args.python,
str(SCRIPT_DIR / "sync_cep_rejection_goods_top_assets.py"),
*common_cfg_args,
"--report-year",
str(year),
"--report-month",
month,
],
)
if not args.skip_compose:
run_step(
8,
total_steps,
[
args.python,
str(SCRIPT_DIR / "compose_rejection_report.py"),
*common_cfg_args,
],
)
final_image = workdir / "assets" / "cep-rejection-analysis" / f"cep-rejection-combined-{year}{month}.png"
print(f"FINAL_IMAGE={final_image}")
print("FLOW18_PIPELINE_OK")
finally:
if tmp_config_path:
try:
tmp_config_path.unlink()
except FileNotFoundError:
pass
if __name__ == "__main__":
main()
vip-report/scripts/runtime_config.py 0 → 100644
View file @ 4b9160b1
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
from typing import Any, Optional
import yaml
DEFAULT_CONFIG_PATH = Path(r"D:\projects\skills\vip-report\config.yaml")
DEFAULT_CONFIG_TABLE = "cepconfiguration"
DEFAULT_CONFIG_KEY = "vip-report"
SCRIPT_DIR = Path(__file__).resolve().parent
for dep_dir in [SCRIPT_DIR / ".pydeps", SCRIPT_DIR.parent / ".pydeps", Path.cwd() / ".pydeps"]:
if dep_dir.exists() and str(dep_dir) not in sys.path:
sys.path.insert(0, str(dep_dir))
def _import_pymysql():
try:
import pymysql # type: ignore
except Exception as exc:
raise RuntimeError(
"Missing dependency pymysql. Please install it in ShadowBot Python, or copy a .pydeps folder containing pymysql."
) from exc
return pymysql
def add_config_source_args(parser: argparse.ArgumentParser, *, default_config: Optional[Path] = None) -> None:
parser.add_argument("--config", default=str(default_config or DEFAULT_CONFIG_PATH))
# Default to DB so automation can run without carrying local config files.
parser.add_argument("--config-source", choices=["yaml", "db"], default="db")
parser.add_argument("--config-table", default=DEFAULT_CONFIG_TABLE)
parser.add_argument("--config-key", default=DEFAULT_CONFIG_KEY)
parser.add_argument("--mysql-host", default="192.168.138.182")
parser.add_argument("--mysql-port", type=int, default=3306)
parser.add_argument("--mysql-user", default="ceptest")
parser.add_argument("--mysql-password", default="ck@123456")
parser.add_argument("--mysql-database", default="ckc_cep_db_test")
def mysql_conn_kwargs_from_args(args: argparse.Namespace) -> dict[str, Any]:
return {
"host": args.mysql_host,
"port": int(args.mysql_port),
"user": args.mysql_user,
"password": args.mysql_password,
"database": args.mysql_database,
"charset": "utf8mb4",
}
def load_runtime_config(args: argparse.Namespace, *, default_config: Optional[Path] = None) -> dict[str, Any]:
if args.config_source == "yaml":
path = Path(args.config or default_config or DEFAULT_CONFIG_PATH).resolve()
return yaml.safe_load(path.read_text(encoding="utf-8"))
return load_db_config(args)
def load_db_config(args: argparse.Namespace) -> dict[str, Any]:
pymysql = _import_pymysql()
table = str(args.config_table or DEFAULT_CONFIG_TABLE).strip()
key = str(args.config_key or DEFAULT_CONFIG_KEY).strip()
conn_args = mysql_conn_kwargs_from_args(args)
conn_args["cursorclass"] = pymysql.cursors.DictCursor
conn = pymysql.connect(**conn_args)
try:
with conn.cursor() as cur:
cur.execute(
"""
SELECT COLUMN_NAME
FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME=%s
""",
(table,),
)
cols = {str(r["COLUMN_NAME"]).lower() for r in (cur.fetchall() or [])}
if {"config_key", "config_json"}.issubset(cols):
sql = f"SELECT config_json AS cfg FROM `{table}` WHERE config_key=%s ORDER BY id DESC LIMIT 1"
elif {"key", "value"}.issubset(cols):
sql = f"SELECT `value` AS cfg FROM `{table}` WHERE `key`=%s ORDER BY id DESC LIMIT 1"
else:
raise RuntimeError(
f"Unsupported config table schema for `{table}`. "
f"Need columns (config_key,config_json) or (key,value), actual={sorted(cols)}"
)
cur.execute(sql, (key,))
row = cur.fetchone()
finally:
conn.close()
if not row:
raise RuntimeError(f"No config found in `{table}` for config_key={key!r}")
raw = row.get("cfg")
if raw is None:
raise RuntimeError(f"`{table}` row found but config_json is NULL for config_key={key!r}")
if isinstance(raw, (dict, list)):
return raw
text = str(raw).strip()
if not text:
raise RuntimeError(f"`{table}` row found but config_json is empty for config_key={key!r}")
try:
return json.loads(text)
except Exception:
return yaml.safe_load(text)
vip-report/scripts/sync_cep_live_goods_effect_pivot_assets.py 0 → 100644
View file @ 4b9160b1
from __future__ import annotations
import argparse
import json
import os
import re
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any
_MPL_CACHE = Path(__file__).resolve().parents[1] / ".mpl-cache"
_MPL_CACHE.mkdir(parents=True, exist_ok=True)
os.environ.setdefault("MPLCONFIGDIR", str(_MPL_CACHE))
SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPT_DIR))
for dep_dir in [SCRIPT_DIR / ".pydeps", SCRIPT_DIR.parent / ".pydeps", Path.cwd() / ".pydeps"]:
if dep_dir.exists() and str(dep_dir) not in sys.path:
sys.path.insert(0, str(dep_dir))
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
from matplotlib.lines import Line2D
import pymysql
import yaml
from runtime_config import add_config_source_args, load_runtime_config
TABLE_NAME = "cep_live_goods_effect"
@dataclass(frozen=True)
class ReportMonth:
year: int
month: int
@property
def ym(self) -> str:
return f"{self.year}-{self.month:02d}"
def parse_month_number(raw: Any) -> int:
if raw is None:
return 1
if isinstance(raw, int):
return min(12, max(1, raw))
text = str(raw).strip()
m = re.search(r"(1[0-2]|0?[1-9])", text)
if m:
return int(m.group(1))
return 1
def resolve_report_month(args: argparse.Namespace, config: dict[str, Any]) -> ReportMonth:
report_cfg = config.get("report", {})
year = int(args.report_year or report_cfg.get("year", 2026))
month = parse_month_number(args.report_month or report_cfg.get("month_cn", "1"))
return ReportMonth(year=year, month=month)
def pick_font() -> str:
candidates = [
r"C:\Windows\Fonts\msyh.ttc",
r"C:\Windows\Fonts\simhei.ttf",
r"C:\Windows\Fonts\simsun.ttc",
]
for p in candidates:
if Path(p).exists():
return p
return "sans-serif"
def fetch_category_data(conn: pymysql.connections.Connection, month: ReportMonth) -> list[dict[str, Any]]:
sql = f"""
SELECT
COALESCE(NULLIF(TRIM(category_level_1), ''), '未分类') AS level1,
COALESCE(NULLIF(TRIM(category_level_3), ''), '未分类') AS level3,
SUM(IFNULL(item_count_total, 0)) AS item_count
FROM `{TABLE_NAME}`
WHERE data_month = %s
GROUP BY level1, level3
HAVING SUM(IFNULL(item_count_total, 0)) > 0
"""
with conn.cursor() as cur:
cur.execute(sql, (month.ym,))
return cur.fetchall() or []
def pct(v: float, total: float) -> float:
if total <= 0:
return 0.0
return v / total
def fmt_pct(v: float) -> str:
return f"{round(v * 100):.0f}%"
def render_chart(rows: list[dict[str, Any]], output_path: Path) -> dict[str, Any]:
if not rows:
raise RuntimeError("No data found for chart rendering.")
level1_totals: dict[str, float] = {}
for r in rows:
l1 = str(r["level1"])
level1_totals[l1] = level1_totals.get(l1, 0.0) + float(r["item_count"] or 0)
level1_sorted = sorted(level1_totals.items(), key=lambda x: x[1], reverse=True)
l1_labels = [x[0] for x in level1_sorted]
l1_values = [x[1] for x in level1_sorted]
l1_total = sum(l1_values)
top_l1 = l1_labels[0]
top_l1_share = pct(l1_values[0], l1_total)
# 二级/三级分类按全量数据汇总（不限制在一级最大类），避免遗漏如挂件等品类。
level3_totals: dict[str, float] = {}
for r in rows:
l3 = str(r["level3"])
level3_totals[l3] = level3_totals.get(l3, 0.0) + float(r["item_count"] or 0)
level3_sorted = sorted(level3_totals.items(), key=lambda x: x[1], reverse=True)
l3_total = sum(v for _, v in level3_sorted)
main_l3 = level3_sorted[:5]
tail_l3 = level3_sorted[5:]
font_name = pick_font()
plt.rcParams["font.sans-serif"] = [font_name, "Microsoft YaHei", "SimHei", "Arial Unicode MS"]
plt.rcParams["axes.unicode_minus"] = False
fig = plt.figure(figsize=(12.8, 5.3), dpi=200, facecolor="#e7e7e7")
ax1 = fig.add_axes([0.05, 0.14, 0.27, 0.72], facecolor="#e7e7e7")
ax2 = fig.add_axes([0.31, 0.14, 0.33, 0.72], facecolor="#e7e7e7")
ax3 = fig.add_axes([0.72, 0.19, 0.24, 0.56], facecolor="#e7e7e7")
c1 = ["#6f76e8", "#e38bcb", "#8db2f5", "#9a8fe8", "#f2a66a", "#8fd3bd", "#c4c4c4"]
ax1.pie(
l1_values,
labels=[f"{k}\n{fmt_pct(pct(v, l1_total))}" if pct(v, l1_total) >= 0.03 else "" for k, v in level1_sorted],
colors=c1[: len(l1_values)],
startangle=90,
counterclock=False,
wedgeprops={"width": 0.28, "edgecolor": "#f0f0f0", "linewidth": 0.6},
labeldistance=1.12,
textprops={"fontsize": 9},
)
ax1.set_title("一级分类", fontsize=12, pad=8)
ax1.text(0, 0.05, top_l1, ha="center", va="center", fontsize=11, color="#333333")
ax1.text(0, -0.12, fmt_pct(top_l1_share), ha="center", va="center", fontsize=11, color="#333333")
ax1.axis("equal")
c2 = ["#4e83c3", "#d66566", "#8eb4e3", "#74b5ad", "#bda6e6", "#f2b38f", "#d8c87d", "#9cc9dd"]
l3_labels = [k for k, _ in main_l3]
l3_values = [v for _, v in main_l3]
labels = [f"{k} {fmt_pct(pct(v, l3_total))}" for k, v in main_l3]
ax2.pie(
l3_values,
labels=labels,
colors=c2[: len(l3_values)],
startangle=270,
counterclock=False,
wedgeprops={"edgecolor": "#f0f0f0", "linewidth": 0.6},
labeldistance=1.05,
textprops={"fontsize": 9},
)
ax2.set_title("三级分类", fontsize=12, pad=8)
ax2.axis("equal")
if tail_l3:
tail_items = tail_l3[:10]
colors_tail = ["#6b8b37", "#5f2d7a", "#2e6ea3", "#ba5a2a", "#7f7f7f", "#a57ab5", "#6e9b8a", "#b38f5a", "#7092be", "#c0504d"]
n = len(tail_items)
for i, (lab, val) in enumerate(tail_items):
y = n - 1 - i
ax3.barh([y], [1.0], color=colors_tail[i % len(colors_tail)], height=0.78)
ax3.text(1.06, y, f"{lab}, {fmt_pct(pct(val, l3_total))}", va="center", fontsize=8, color="#333333")
ax3.set_xlim(0, 1.95)
ax3.set_ylim(-0.8, max(0.8, n - 0.2))
ax3.axis("off")
fig.add_artist(Line2D([0.62, 0.72], [0.47, 0.58], transform=fig.transFigure, color="#9f9f9f", linewidth=1))
fig.add_artist(Line2D([0.62, 0.72], [0.41, 0.30], transform=fig.transFigure, color="#9f9f9f", linewidth=1))
else:
ax3.axis("off")
output_path.parent.mkdir(parents=True, exist_ok=True)
fig.savefig(output_path, dpi=200, facecolor=fig.get_facecolor(), bbox_inches="tight")
plt.close(fig)
return {
"top_level1": top_l1,
"top_level1_count": int(l1_values[0]),
"top_level1_share": round(top_l1_share, 4),
"level1_count": len(level1_sorted),
"level3_count": len(level3_sorted),
"total_item_count": int(l1_total),
"main_level3": [{"name": k, "value": int(v)} for k, v in main_l3],
"tail_level3": [{"name": k, "value": int(v)} for k, v in tail_l3],
}
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Generate goods-effect pivot chart from cep_live_goods_effect.")
add_config_source_args(parser, default_config=Path(r"D:\projects\skills\vip-report\config.yaml"))
parser.add_argument("--report-year", type=int, default=0, help="Override report year")
parser.add_argument("--report-month", default="", help="Override report month")
return parser.parse_args()
def main() -> None:
args = parse_args()
config_path = Path(args.config or r"D:\projects\skills\vip-report\config.yaml").resolve()
config = load_runtime_config(args, default_config=config_path)
report_month = resolve_report_month(args, config)
mysql_cfg = config["mysql"]
mysql_user = mysql_cfg.get("username") or mysql_cfg.get("user")
conn = pymysql.connect(
host=mysql_cfg["host"],
port=int(mysql_cfg["port"]),
user=mysql_user,
password=mysql_cfg["password"],
database=mysql_cfg["database"],
charset="utf8mb4",
cursorclass=pymysql.cursors.DictCursor,
)
try:
rows = fetch_category_data(conn, report_month)
finally:
conn.close()
vip_workdir = Path(config["paths"]["workdir"]).resolve()
asset_dir = vip_workdir / "assets" / "cep-live-goods-effect"
data_dir = vip_workdir / "data" / "cep-live-goods-effect"
image_path = asset_dir / "cep-live-goods-effect-pivot.png"
operations_path = vip_workdir / "render-ops.cep-live-goods-effect-pivot.live.json"
manifest_path = data_dir / "cep-live-goods-effect-pivot.live.json"
metrics = render_chart(rows, image_path)
operations = {"replace_text": [], "replace_images": []}
operations_path.write_text(json.dumps(operations, ensure_ascii=False, indent=2), encoding="utf-8")
manifest = {
"source": {"table": TABLE_NAME, "month": report_month.ym, "config_path": str(config_path)},
"metrics": metrics,
"asset_path": str(image_path),
"operations_path": str(operations_path),
}
manifest_path.parent.mkdir(parents=True, exist_ok=True)
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
print(
json.dumps(
{"image_path": str(image_path), "manifest_path": str(manifest_path), "operations_path": str(operations_path)},
ensure_ascii=False,
)
)
if __name__ == "__main__":
main()
vip-report/scripts/sync_cep_live_goods_effect_top5_assets.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/sync_cep_live_session_assets.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/sync_cep_live_session_trend_assets.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/sync_cep_rejection_goods_top_assets.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/sync_cep_rejection_lfl_assets.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/sync_cep_site_traffic_entry_assets.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/sync_cep_traffic_assets.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/sync_cep_traffic_summary_assets.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/sync_compass_rejection_analysis_capture.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/sync_compass_rejection_goods_download.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/sync_compass_rejection_scene_download.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/sync_compass_traffic_source_download.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/sync_inventory_monthly_assets.py
View file @ 4b9160b1
...@@ -702,8 +702,8 @@ def build_specs(report_month: str, report_year: int, compare_year: int) -> dict[ ...@@ -702,8 +702,8 @@ def build_specs(report_month: str, report_year: int, compare_year: int) -> dict[
{ {
"slide_code": "S05", "slide_code": "S05",
"slide": 5, "slide": 5,
"shape_name": "鍥剧墖 1", "shape_name": "Picture 1",
"shape_id": 2, "shape_id": 5,
"asset_name": "s05_bags_top", "asset_name": "s05_bags_top",
"capture_id": "bags_full", "capture_id": "bags_full",
"crop": {"left": 8, "top": 70, "width": 1148, "height": 250}, "crop": {"left": 8, "top": 70, "width": 1148, "height": 250},
...@@ -714,8 +714,8 @@ def build_specs(report_month: str, report_year: int, compare_year: int) -> dict[ ...@@ -714,8 +714,8 @@ def build_specs(report_month: str, report_year: int, compare_year: int) -> dict[
{ {
"slide_code": "S05", "slide_code": "S05",
"slide": 5, "slide": 5,
"shape_name": "鍥剧墖 2", "shape_name": "Picture 2",
"shape_id": 3, "shape_id": 7,
"asset_name": "s05_bags_bottom", "asset_name": "s05_bags_bottom",
"capture_id": "bags_full", "capture_id": "bags_full",
"crop": {"left": 8, "top": 1325, "width": 1148, "height": 395}, "crop": {"left": 8, "top": 1325, "width": 1148, "height": 395},
...@@ -2780,17 +2780,35 @@ def ensure_browser_session( ...@@ -2780,17 +2780,35 @@ def ensure_browser_session(
browser_config_path: Path | None = None, browser_config_path: Path | None = None,
) -> None: ) -> None:
"""Helper.""" """Helper."""
session_available = False
try: try:
run_playwright( run_playwright(
["--session", session, "tab-list"], ["--session", session, "tab-list"],
cwd=cwd, cwd=cwd,
timeout=remaining_timeout(15, deadline=deadline, action="checking existing browser session"), timeout=remaining_timeout(45, deadline=deadline, action="checking existing browser session"),
) )
return session_available = True
except subprocess.TimeoutExpired:
session_available = False
except subprocess.CalledProcessError as error: except subprocess.CalledProcessError as error:
combined = f"{error.stdout}\n{error.stderr}" combined = f"{error.stdout}\n{error.stderr}"
if "is not open" not in combined: if "is not open" not in combined:
raise raise
session_available = False
if session_available:
return
# Cleanup stale session state before opening a fresh browser session.
try:
run_playwright(
["--session", session, "close"],
cwd=cwd,
timeout=remaining_timeout(30, deadline=deadline, action="closing stale browser session"),
)
except Exception:
pass
args = ["--session", session, "open"] args = ["--session", session, "open"]
resolved_browser_config_path = browser_config_path or PLAYWRIGHT_CLI_CONFIG_PATH resolved_browser_config_path = browser_config_path or PLAYWRIGHT_CLI_CONFIG_PATH
if resolved_browser_config_path is not None: if resolved_browser_config_path is not None:
...@@ -2799,7 +2817,7 @@ def ensure_browser_session( ...@@ -2799,7 +2817,7 @@ def ensure_browser_session(
run_playwright( run_playwright(
args, args,
cwd=cwd, cwd=cwd,
timeout=remaining_timeout(60, deadline=deadline, action="opening the browser session"), timeout=remaining_timeout(90, deadline=deadline, action="opening the browser session"),
) )
......
vip-report/scripts/sync_top_products_assets.py
View file @ 4b9160b1
...@@ -69,7 +69,7 @@ SLIDE_SPECS: dict[str, dict[str, Any]] = { ...@@ -69,7 +69,7 @@ SLIDE_SPECS: dict[str, dict[str, Any]] = {
"asset_name": "s09_top_products_chart", "asset_name": "s09_top_products_chart",
"raw_screenshot_name": "top-products-bags.png", "raw_screenshot_name": "top-products-bags.png",
"resize_to": {"width": 1034, "height": 587}, "resize_to": {"width": 1034, "height": 587},
"crop": {"left": 0, "top": 480, "width": 900, "height": 1900}, "crop": {"left": 0, "top": 480, "width": 1100, "height": 1900},
"note": "S09 main Tableau chart area.", "note": "S09 main Tableau chart area.",
"product_slots": [ "product_slots": [
{"rank": 1, "shape_id": 4, "shape_name": "Picture 2", "canvas_width": 707, "canvas_height": 500}, {"rank": 1, "shape_id": 4, "shape_name": "Picture 2", "canvas_width": 707, "canvas_height": 500},
...@@ -94,7 +94,7 @@ SLIDE_SPECS: dict[str, dict[str, Any]] = { ...@@ -94,7 +94,7 @@ SLIDE_SPECS: dict[str, dict[str, Any]] = {
"asset_name": "s10_top_products_chart", "asset_name": "s10_top_products_chart",
"raw_screenshot_name": "top-products-shoes.png", "raw_screenshot_name": "top-products-shoes.png",
"resize_to": {"width": 1039, "height": 585}, "resize_to": {"width": 1039, "height": 585},
"crop": {"left": 0, "top": 480, "width": 900, "height": 1900}, "crop": {"left": 0, "top": 480, "width": 1100, "height": 1900},
"note": "S10 main Tableau chart area.", "note": "S10 main Tableau chart area.",
"product_slots": [ "product_slots": [
{"rank": 1, "shape_id": 3, "shape_name": "Picture 2", "canvas_width": 708, "canvas_height": 500}, {"rank": 1, "shape_id": 3, "shape_name": "Picture 2", "canvas_width": 708, "canvas_height": 500},
...@@ -2299,14 +2299,14 @@ def render_sql_top_products_chart(source_payload: dict[str, Any], target: Path) ...@@ -2299,14 +2299,14 @@ def render_sql_top_products_chart(source_payload: dict[str, Any], target: Path)
top_sales_percentage = sum(sales_values) / total_sales * 100 top_sales_percentage = sum(sales_values) / total_sales * 100
figure = plt.figure(figsize=(10.34, 5.87), dpi=100, facecolor="white") figure = plt.figure(figsize=(10.34, 5.87), dpi=100, facecolor="white")
grid = GridSpec(1, 4, figure=figure, width_ratios=[4.9, 1.8, 1.8, 2.2], wspace=0.08) grid = GridSpec(1, 4, figure=figure, width_ratios=[4.7, 1.8, 2.35, 2.25], wspace=0.14)
ax_qty = figure.add_subplot(grid[0, 0]) ax_qty = figure.add_subplot(grid[0, 0])
ax_qty_pct = figure.add_subplot(grid[0, 1], sharey=ax_qty) ax_qty_pct = figure.add_subplot(grid[0, 1], sharey=ax_qty)
ax_sales_pct = figure.add_subplot(grid[0, 2], sharey=ax_qty) ax_sales_pct = figure.add_subplot(grid[0, 2], sharey=ax_qty)
ax_sales = figure.add_subplot(grid[0, 3], sharey=ax_qty) ax_sales = figure.add_subplot(grid[0, 3], sharey=ax_qty)
y_positions = list(range(len(rows))) y_positions = list(range(len(rows)))
figure.subplots_adjust(left=0.03, right=0.99, top=0.83, bottom=0.05) figure.subplots_adjust(left=0.03, right=0.94, top=0.83, bottom=0.05)
figure.text(0.02, 0.96, "Top Parameter", ha="left", va="bottom", fontsize=10) figure.text(0.02, 0.96, "Top Parameter", ha="left", va="bottom", fontsize=10)
figure.patches.append( figure.patches.append(
Rectangle((0.02, 0.915), 0.31, 0.034, transform=figure.transFigure, fill=False, linewidth=0.8, edgecolor="#666666") Rectangle((0.02, 0.915), 0.31, 0.034, transform=figure.transFigure, fill=False, linewidth=0.8, edgecolor="#666666")
...@@ -2346,30 +2346,43 @@ def render_sql_top_products_chart(source_payload: dict[str, Any], target: Path) ...@@ -2346,30 +2346,43 @@ def render_sql_top_products_chart(source_payload: dict[str, Any], target: Path)
ax_qty_pct.text(-value - pct_limit * 0.05, index, f"{round(value):.0f}%", fontsize=9, fontweight="bold", ha="right", va="center") ax_qty_pct.text(-value - pct_limit * 0.05, index, f"{round(value):.0f}%", fontsize=9, fontweight="bold", ha="right", va="center")
style_chart_axis(ax_qty_pct) style_chart_axis(ax_qty_pct)
ax_sales_pct.set_xlim(0, pct_limit * 1.18) sales_pct_axis_limit = pct_limit * 1.45
ax_sales_pct.set_xlim(0, sales_pct_axis_limit)
ax_sales_pct.barh(y_positions, sales_percentages, height=0.75, color="#8db9db") ax_sales_pct.barh(y_positions, sales_percentages, height=0.75, color="#8db9db")
ax_sales_pct.axvline(0, color="#d6d6d6", linewidth=0.8) ax_sales_pct.axvline(0, color="#d6d6d6", linewidth=0.8)
ax_sales_pct.set_title(f"Sales%\n{round(top_sales_percentage):.0f}%", fontsize=12, fontweight="bold", pad=24) ax_sales_pct.set_title(f"Sales%\n{round(top_sales_percentage):.0f}%", fontsize=12, fontweight="bold", pad=24)
for index, value in enumerate(sales_percentages): for index, value in enumerate(sales_percentages):
ax_sales_pct.text(value + pct_limit * 0.05, index, f"{round(value):.0f}%", fontsize=9, fontweight="bold", ha="left", va="center") ax_sales_pct.text(
min(value + pct_limit * 0.045, sales_pct_axis_limit * 0.96),
index,
f"{round(value):.0f}%",
fontsize=9,
fontweight="bold",
ha="left",
va="center",
clip_on=False,
)
style_chart_axis(ax_sales_pct) style_chart_axis(ax_sales_pct)
sales_title_x = (ax_qty_pct.get_position().x0 + ax_sales_pct.get_position().x1) / 2 sales_title_x = (ax_qty_pct.get_position().x0 + ax_sales_pct.get_position().x1) / 2
figure.text(sales_title_x, 0.965, "TOP N%", ha="center", va="bottom", fontsize=13, fontweight="bold") figure.text(sales_title_x, 0.965, "TOP N%", ha="center", va="bottom", fontsize=13, fontweight="bold")
max_soh = max(soh_values) or 1.0 max_soh = max(soh_values) or 1.0
ax_sales.set_xlim(0, max_soh * 1.22) ax_sales.set_xlim(0, max_soh * 1.55)
ax_sales.barh(y_positions, soh_values, height=0.75, color="#a7a7a7") ax_sales.barh(y_positions, soh_values, height=0.75, color="#a7a7a7")
ax_sales.axvline(0, color="#d6d6d6", linewidth=0.8) ax_sales.axvline(0, color="#d6d6d6", linewidth=0.8)
for index, value in enumerate(soh_values): for index, value in enumerate(soh_values):
is_long_bar = value >= max_soh * 0.62
label_x = value - max_soh * 0.035 if is_long_bar else value + max_soh * 0.035
ax_sales.text( ax_sales.text(
value + max_soh * 0.02, label_x,
index, index,
format_chart_number(value), format_chart_number(value),
fontsize=9, fontsize=9,
fontweight="bold", fontweight="bold",
ha="left", ha="right" if is_long_bar else "left",
va="center", va="center",
clip_on=True,
) )
ax_sales.text(0, len(rows) - 0.08, "soh", fontsize=9, ha="left", va="bottom") ax_sales.text(0, len(rows) - 0.08, "soh", fontsize=9, ha="left", va="bottom")
style_chart_axis(ax_sales) style_chart_axis(ax_sales)
......
vip-report/scripts/sync_vis_live_session_downloads.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/sync_vis_member_channel_trend_capture.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/sync_vis_member_consumption_capture.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/sync_vis_member_overview_capture.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/sync_vis_vendor_goods_effect_download.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
vip-report/scripts/upsert_cepconfiguration.py 0 → 100644
View file @ 4b9160b1
This diff is collapsed.
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment