huanghai/aiData
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

'commit'

Browse Source
This commit is contained in:
HuangHai
2026-01-15 10:06:08 +08:00
parent 98e890a693
commit ccac3e889c
13 changed files with 35 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
Apps/XinDianTu/Report/Excel/新电途_export_20260113_120046.xlsx
View File

Binary file not shown.

505
Apps/XinDianTu/Report/Export.py
View File

@@ -1,505 +0,0 @@
import asyncio
import json
import logging
import os
import re
import sys
from datetime import datetime
from decimal import Decimal
from typing import Any, Dict, List, Optional, Set
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)
project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
if project_root not in sys.path:
sys.path.append(project_root)
from Util import Win32Patch
Win32Patch.patch()
from Config.Config import DB_URL
from DbKit.Db import Db
try:
from openpyxl import Workbook
from openpyxl.styles import Alignment, Border, Font, PatternFill, Side
from openpyxl.utils import get_column_letter
except Exception as e:
raise RuntimeError("缺少 openpyxl 依赖，请先安装：pip install openpyxl") from e
# --- 样式定义 ---
# 颜色
HEADER_FILL_COLOR = "4472C4" # 专业蓝 (表头背景)
HEADER_FONT_COLOR = "FFFFFF" # 白色 (表头文字)
LABEL_FILL_COLOR = "D9E1F2" # 浅蓝 (Key标签背景，更柔和)
SECTION_FILL_COLOR = "DDEBF7" # 极浅蓝 (小节标题背景)
BORDER_COLOR = "BFBFBF" # 边框颜色
OUTER_BORDER_COLOR = "4472C4" # 外边框颜色
# 字体
FONT_NAME = "微软雅黑"
TITLE_FONT = Font(name=FONT_NAME, size=20, bold=True, color="4472C4") # 主标题加大
SUBTITLE_FONT = Font(name=FONT_NAME, size=12, bold=True, color="203764") # 小节标题加深色
HEADER_FONT = Font(name=FONT_NAME, size=11, bold=True, color=HEADER_FONT_COLOR)
LABEL_FONT = Font(name=FONT_NAME, size=10, bold=True, color="333333") # 标签文字深灰
DATA_FONT = Font(name=FONT_NAME, size=10)
# 边框
THIN_SIDE = Side(style="thin", color=BORDER_COLOR)
MEDIUM_SIDE = Side(style="medium", color=OUTER_BORDER_COLOR)
BORDER_ALL_THIN = Border(left=THIN_SIDE, right=THIN_SIDE, top=THIN_SIDE, bottom=THIN_SIDE)
# 填充
HEADER_FILL = PatternFill("solid", fgColor=HEADER_FILL_COLOR)
LABEL_FILL = PatternFill("solid", fgColor=LABEL_FILL_COLOR)
SECTION_FILL = PatternFill("solid", fgColor=SECTION_FILL_COLOR)
INVALID_SHEET_CHARS_RE = re.compile(r"[:\\/?*\[\]]")
def _safe_excel_sheet_title(raw_title: str, existing_titles: Set[str]) -> str:
title = (raw_title or "").strip()
if not title:
title = "未命名场站"
title = INVALID_SHEET_CHARS_RE.sub("_", title)
title = title.replace("\n", " ").replace("\r", " ").strip()
if len(title) > 31:
title = title[:31].rstrip()
if not title:
title = "未命名场站"
if title not in existing_titles:
existing_titles.add(title)
return title
base = title[:28].rstrip() if len(title) > 28 else title
idx = 2
while True:
candidate = f"{base}_{idx}"
if len(candidate) > 31:
candidate = candidate[:31]
if candidate not in existing_titles:
existing_titles.add(candidate)
return candidate
idx += 1
def _to_float(value):
if value is None:
return None
if isinstance(value, Decimal):
return float(value)
return value
def _parse_schedule(schedule_json: Any) -> List[Dict[str, Any]]:
if not schedule_json:
return []
if isinstance(schedule_json, (dict, list)):
return schedule_json if isinstance(schedule_json, list) else [schedule_json]
if isinstance(schedule_json, (bytes, bytearray)):
try:
schedule_json = schedule_json.decode("utf-8", errors="ignore")
except Exception:
schedule_json = str(schedule_json)
if not isinstance(schedule_json, str):
return []
try:
data = json.loads(schedule_json)
if isinstance(data, list):
return data
if isinstance(data, dict):
return [data]
return []
except Exception:
return []
def _apply_table_header_style(ws, row_idx: int, col_start: int, col_end: int):
alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
# Header row with top/bottom medium border
border = Border(left=THIN_SIDE, right=THIN_SIDE, top=MEDIUM_SIDE, bottom=MEDIUM_SIDE)
for col in range(col_start, col_end + 1):
cell = ws.cell(row=row_idx, column=col)
cell.fill = HEADER_FILL
cell.font = HEADER_FONT
cell.alignment = alignment
cell.border = border
ws.row_dimensions[row_idx].height = 28 # Increased header height
def _apply_table_cell_style(ws, row_idx: int, col_start: int, col_end: int, center_cols: List[int] = None):
for col in range(col_start, col_end + 1):
cell = ws.cell(row=row_idx, column=col)
cell.font = DATA_FONT
is_center = center_cols and col in center_cols
cell.alignment = Alignment(horizontal="center" if is_center else "left", vertical="center", wrap_text=True)
cell.border = BORDER_ALL_THIN
ws.row_dimensions[row_idx].height = 22 # Comfortable row height
def _style_info_key_value(ws, row, col_key, col_val, merge_val_cols=0):
# Key Style
c_key = ws.cell(row=row, column=col_key)
c_key.font = LABEL_FONT
c_key.alignment = Alignment(horizontal="right", vertical="center")
c_key.fill = LABEL_FILL
c_key.border = BORDER_ALL_THIN
# Value Style
c_val = ws.cell(row=row, column=col_val)
c_val.font = DATA_FONT
c_val.alignment = Alignment(horizontal="left", vertical="center", wrap_text=True)
c_val.border = BORDER_ALL_THIN
if merge_val_cols > 0:
ws.merge_cells(start_row=row, start_column=col_val, end_row=row, end_column=col_val + merge_val_cols)
# Apply border to merged cells (required for openpyxl)
for c_offset in range(1, merge_val_cols + 1):
c_merged = ws.cell(row=row, column=col_val + c_offset)
c_merged.border = BORDER_ALL_THIN
def _set_column_widths(ws, widths: dict[int, float]):
for col_idx, width in widths.items():
ws.column_dimensions[get_column_letter(col_idx)].width = width
class DorisExcelExporter:
def __init__(self, db_url: str):
self.db = Db(db_url=db_url)
async def init(self):
await self.db.init_db()
async def close(self):
try:
await self.db.close()
finally:
if getattr(self.db, "engine", None):
await self.db.engine.dispose()
async def fetch_current_station_rows(self) -> List[Dict[str, Any]]:
sql = """
SELECT
p.station_hash,
p.station_name,
p.address,
p.coord_x,
p.coord_y,
s.total_piles AS total_guns,
s.free_piles AS free_guns,
s.piles_detail_json,
s.current_price,
s.pro_price,
s.parking_info,
s.distance,
s.valid_start_time AS status_update_time,
pr.schedule_json,
pr.valid_start_time AS schedule_update_time
FROM t_station_profile_scd p
LEFT JOIN t_station_status_scd s
ON p.station_hash = s.station_hash AND s.is_current = 1
LEFT JOIN t_station_price_schedule_scd pr
ON p.station_hash = pr.station_hash AND pr.is_current = 1
WHERE p.is_current = 1
ORDER BY p.station_name
"""
return await self.db.find(sql)
def _build_station_sheet(ws, row: Dict[str, Any]):
station_name = row.get("station_name") or ""
address = row.get("address") or ""
coord_x = row.get("coord_x")
coord_y = row.get("coord_y")
total_guns = row.get("total_guns")
free_guns = row.get("free_guns")
current_price = _to_float(row.get("current_price"))
pro_price = _to_float(row.get("pro_price"))
parking_info = row.get("parking_info")
distance = row.get("distance")
status_update_time = row.get("status_update_time")
schedule_update_time = row.get("schedule_update_time")
schedule_items = _parse_schedule(row.get("schedule_json"))
piles_detail = _parse_schedule(row.get("piles_detail_json")) # Reuse the same parser
for i in range(1, len(schedule_items)):
curr = schedule_items[i]
prev = schedule_items[i - 1]
if not isinstance(curr, dict) or not isinstance(prev, dict):
continue
for k in ("price_kwh", "electric_fee_kwh", "service_fee_kwh", "price", "electric_fee", "service_fee"):
if curr.get(k) is None and prev.get(k) is not None:
curr[k] = prev[k]
ws.merge_cells(start_row=1, start_column=1, end_row=1, end_column=6)
ws.row_dimensions[1].height = 40 # Increased title height
title_cell = ws.cell(row=1, column=1, value=station_name or "场站")
title_cell.font = TITLE_FONT
title_cell.alignment = Alignment(horizontal="center", vertical="center")
# Title border (Optional: bottom medium)
for c in range(1, 7):
ws.cell(row=1, column=c).border = Border(bottom=MEDIUM_SIDE)
ws.sheet_properties.tabColor = "4472C4" # Tab color matches theme
# Layout using _style_info_key_value
# Row 3: 位置 | Address
_style_info_key_value(ws, 3, 1, 2, merge_val_cols=4)
ws.cell(row=3, column=1, value="位置")
ws.cell(row=3, column=2, value=address)
# Row 4: 经纬度 | Coords | 距离 | Distance
coord_text = f"{coord_x}, {coord_y}" if coord_x is not None else ""
_style_info_key_value(ws, 4, 1, 2, merge_val_cols=1) # 经纬度 occupies col 2-3
ws.cell(row=4, column=1, value="经纬度")
ws.cell(row=4, column=2, value=coord_text)
_style_info_key_value(ws, 4, 4, 5, merge_val_cols=1) # 距离 occupies col 5-6
ws.cell(row=4, column=4, value="距离")
ws.cell(row=4, column=5, value=distance)
# Row 5: 空闲 | val | 总枪 | val | 停车 | val (Parking info needs more space, let's adjust)
# Original: Free(1,2), Total(3,4), Parking(5,6) - Parking info is long usually.
# New Layout: Free(1,2), Total(3,4), Parking(5,6) is fine if col 6 is wide.
_style_info_key_value(ws, 5, 1, 2)
ws.cell(row=5, column=1, value="空闲枪数")
ws.cell(row=5, column=2, value=free_guns)
_style_info_key_value(ws, 5, 3, 4)
ws.cell(row=5, column=3, value="总枪数")
ws.cell(row=5, column=4, value=total_guns)
_style_info_key_value(ws, 5, 5, 6)
ws.cell(row=5, column=5, value="停车信息")
ws.cell(row=5, column=6, value=parking_info)
# Row 6: Price | val | Pro Price | val | Update Time | val
_style_info_key_value(ws, 6, 1, 2)
ws.cell(row=6, column=1, value="当前价格")
ws.cell(row=6, column=2, value=current_price)
_style_info_key_value(ws, 6, 3, 4)
ws.cell(row=6, column=3, value="PRO会员价")
ws.cell(row=6, column=4, value=pro_price)
_style_info_key_value(ws, 6, 5, 6)
ws.cell(row=6, column=5, value="更新时间")
ws.cell(row=6, column=6, value=status_update_time)
# Adjust row heights for info section
for r in range(3, 7):
ws.row_dimensions[r].height = 24
# 价格表部分
schedule_start_row = 8
# Section Title with style
sect_cell = ws.cell(row=schedule_start_row, column=1, value="价格时段表")
sect_cell.font = SUBTITLE_FONT
sect_cell.fill = SECTION_FILL
sect_cell.border = Border(bottom=THIN_SIDE)
ws.merge_cells(start_row=schedule_start_row, start_column=1, end_row=schedule_start_row, end_column=6)
# Apply fill to merged cells
for c in range(2, 7):
ws.cell(row=schedule_start_row, column=c).fill = SECTION_FILL
ws.cell(row=schedule_start_row, column=c).border = Border(bottom=THIN_SIDE)
header_row = schedule_start_row + 1
headers = ["开始", "结束", "电价(元/kWh)", "电费(元/kWh)", "服务费(元/kWh)", "备注"]
for i, h in enumerate(headers, start=1):
ws.cell(row=header_row, column=i, value=h)
_apply_table_header_style(ws, header_row, 1, len(headers))
row_idx = header_row + 1
# Track last valid values for forward filling
last_valid_price = None
last_valid_electric = None
last_valid_service = None
for item in schedule_items:
if not isinstance(item, dict):
continue
start = item.get("start") or item.get("begin") or item.get("start_time") or ""
end = item.get("end") or item.get("finish") or item.get("end_time") or ""
price = _to_float(item.get("price_kwh") if "price_kwh" in item else item.get("price"))
electric_fee = _to_float(item.get("electric_fee_kwh") if "electric_fee_kwh" in item else item.get("electric_fee"))
service_fee = _to_float(item.get("service_fee_kwh") if "service_fee_kwh" in item else item.get("service_fee"))
# Forward fill if None
if price is None and last_valid_price is not None:
price = last_valid_price
if electric_fee is None and last_valid_electric is not None:
electric_fee = last_valid_electric
if service_fee is None and last_valid_service is not None:
service_fee = last_valid_service
# Update last valid values if current is not None
if price is not None:
last_valid_price = price
if electric_fee is not None:
last_valid_electric = electric_fee
if service_fee is not None:
last_valid_service = service_fee
remark = item.get("type") or item.get("tag") or item.get("period") or ""
ws.cell(row=row_idx, column=1, value=start)
ws.cell(row=row_idx, column=2, value=end)
ws.cell(row=row_idx, column=3, value=price)
ws.cell(row=row_idx, column=4, value=electric_fee)
ws.cell(row=row_idx, column=5, value=service_fee)
ws.cell(row=row_idx, column=6, value=remark)
_apply_table_cell_style(ws, row_idx, 1, len(headers), center_cols=[1, 2, 3, 4, 5])
row_idx += 1
# 充电桩详情部分
piles_start_row = row_idx + 2
# Section Title with style
sect_cell = ws.cell(row=piles_start_row, column=1, value="充电桩详情")
sect_cell.font = SUBTITLE_FONT
sect_cell.fill = SECTION_FILL
sect_cell.border = Border(bottom=THIN_SIDE)
ws.merge_cells(start_row=piles_start_row, start_column=1, end_row=piles_start_row, end_column=6)
# Apply fill to merged cells
for c in range(2, 7):
ws.cell(row=piles_start_row, column=c).fill = SECTION_FILL
ws.cell(row=piles_start_row, column=c).border = Border(bottom=THIN_SIDE)
pile_header_row = piles_start_row + 1
pile_headers = ["编号", "功率", "类型", "状态", "价格(元/kWh)", "备注"]
for i, h in enumerate(pile_headers, start=1):
ws.cell(row=pile_header_row, column=i, value=h)
_apply_table_header_style(ws, pile_header_row, 1, len(pile_headers))
row_idx = pile_header_row + 1
if piles_detail:
for pile in piles_detail:
if not isinstance(pile, dict): continue
ws.cell(row=row_idx, column=1, value=pile.get("pile_no") or pile.get("id") or "")
ws.cell(row=row_idx, column=2, value=pile.get("power") or "")
ws.cell(row=row_idx, column=3, value=pile.get("type") or "")
ws.cell(row=row_idx, column=4, value=pile.get("status_text") or pile.get("status") or "")
ws.cell(row=row_idx, column=5, value=_to_float(pile.get("price")))
ws.cell(row=row_idx, column=6, value=pile.get("remark") or "")
_apply_table_cell_style(ws, row_idx, 1, len(pile_headers), center_cols=[1, 2, 3, 4, 5])
row_idx += 1
else:
ws.cell(row=row_idx, column=1, value="暂无详情数据")
ws.merge_cells(start_row=row_idx, start_column=1, end_row=row_idx, end_column=6)
_apply_table_cell_style(ws, row_idx, 1, len(pile_headers))
row_idx += 1
ws.freeze_panes = ws["A9"]
_set_column_widths(ws, {1: 20, 2: 20, 3: 16, 4: 16, 5: 16, 6: 25})
def _build_summary_sheet(ws, rows: List[Dict[str, Any]]):
ws.title = "场站总览"
ws.sheet_properties.tabColor = "4472C4" # Tab color matches theme
# Row 1: Main Title
title_cell = ws.cell(row=1, column=1, value="新电途充电场站数据总览")
title_cell.font = TITLE_FONT
title_cell.alignment = Alignment(horizontal="center", vertical="center")
ws.merge_cells(start_row=1, start_column=1, end_row=1, end_column=10)
ws.row_dimensions[1].height = 40
# Row 2: Headers
headers = ["场站名称", "地址", "距离", "经度", "纬度", "总枪数", "空闲枪数", "停车信息", "当前价格(元/kWh)", "状态更新时间"]
for i, h in enumerate(headers, start=1):
ws.cell(row=2, column=i, value=h)
_apply_table_header_style(ws, 2, 1, len(headers))
# Enable AutoFilter
ws.auto_filter.ref = f"A2:{get_column_letter(len(headers))}{len(rows) + 2}"
row_idx = 3
for row in rows:
ws.cell(row=row_idx, column=1, value=row.get("station_name"))
ws.cell(row=row_idx, column=2, value=row.get("address"))
ws.cell(row=row_idx, column=3, value=row.get("distance"))
ws.cell(row=row_idx, column=4, value=row.get("coord_x"))
ws.cell(row=row_idx, column=5, value=row.get("coord_y"))
ws.cell(row=row_idx, column=6, value=row.get("total_guns"))
ws.cell(row=row_idx, column=7, value=row.get("free_guns"))
ws.cell(row=row_idx, column=8, value=row.get("parking_info"))
ws.cell(row=row_idx, column=9, value=_to_float(row.get("current_price")))
ws.cell(row=row_idx, column=10, value=row.get("status_update_time"))
_apply_table_cell_style(ws, row_idx, 1, len(headers), center_cols=[3, 4, 5, 6, 7, 9, 10])
row_idx += 1
_set_column_widths(ws, {1: 35, 2: 45, 3: 12, 4: 14, 5: 14, 6: 12, 7: 12, 8: 35, 9: 20, 10: 22})
ws.freeze_panes = ws["A3"]
async def export_excel(output_path: str, *, limit: Optional[int] = None) -> str:
exporter = DorisExcelExporter(db_url=DB_URL)
await exporter.init()
try:
rows = await exporter.fetch_current_station_rows()
finally:
await exporter.close()
if limit is not None:
rows = rows[: max(0, int(limit))]
wb = Workbook()
if not rows:
ws = wb.active
ws.title = "无数据"
ws.cell(row=1, column=1, value="未查询到任何场站数据")
wb.save(output_path)
return output_path
# 1. 创建总览页
summary_ws = wb.active
_build_summary_sheet(summary_ws, rows)
# 2. 为每个场站创建独立页
existing_titles: Set[str] = {"场站总览"}
for row in rows:
sheet_title = _safe_excel_sheet_title(str(row.get("station_name") or ""), existing_titles)
ws = wb.create_sheet(title=sheet_title)
_build_station_sheet(ws, row)
wb.save(output_path)
return output_path
async def main():
# 配置：默认导出所有数据，如果需要测试少量数据，可以将 None 改为数字，如 10
limit = None
# 构建默认输出路径
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
output_filename = f"./Excel/station_export_{ts}.xlsx"
output_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), output_filename)
logger.info(f"开始导出 Excel...")
logger.info(f"目标文件: {output_path}")
if limit is not None:
logger.info(f"限制导出记录数: {limit}")
final_path = await export_excel(output_path, limit=limit)
logger.info(f"Excel 导出完成: {final_path}")
if __name__ == "__main__":
asyncio.run(main())