""" Excel data loader for daily/weekly/monthly report generation. """ import pandas as pd from datetime import datetime, timedelta import re import warnings # Field mapping: Excel column name -> internal field name FIELD_MAP = { '序号': 'seq', '目的国家': 'country', '合同号': 'contract_no', '用户名称/公司': 'customer', '意向车型及数量': 'product_info', '订单总数量': 'order_qty', '负责人': 'owner', '当前状态': 'status', '拟定合同时间': 'contract_date', '跟单天数': 'tracking_days', '定金支付时间': 'deposit_date', '订金认领时间': 'deposit_claim_date', '订单生成时间': 'order_gen_date', '价格评审时间': 'price_review_date', '合同评审时间': 'contract_review_date', '合同提交盖章申请时间': 'seal_apply_date', '合同盖章时间': 'seal_date', '车辆下线入库状态': 'inventory_status', '尾款支付时间': 'final_pay_date', '尾款认领时间': 'final_claim_date', '智慧关务信息维护': 'customs_date', '许可证办理时间': 'license_date', '车辆发运时间': 'ship_date', '预计开票时间': 'invoice_date', '今日进度更新': 'progress_update', '是否更新': 'is_updated', '支持需求': 'support_request', '4月交付': 'deliver_apr', '5月预测': 'forecast_may', } STATUS_ORDER = ['A', 'B', 'C', 'D', 'E', 'F'] STATUS_LABELS = { 'A': '合同拟定中', 'B': '已锁定合同待付订金', 'C': '已付订金待生产', 'D': '已生产待付尾款', 'E': '已付尾款待发运', 'F': '已发运', } def _normalize_status(val): """Extract status code A-F from status string.""" if pd.isna(val): return None s = str(val).strip() # Match pattern like "A（合同拟定中）" or "A" m = re.match(r'^([A-F])', s) if m: return m.group(1) return None def _parse_date(val): """Parse various date formats.""" if pd.isna(val): return None if isinstance(val, datetime): return val for fmt in ('%Y-%m-%d', '%Y/%m/%d', '%Y年%m月%d日'): try: return datetime.strptime(str(val).strip(), fmt) except ValueError: continue return None def _sheet_name_for_date(date: datetime) -> str: """Convert datetime to expected sheet name.""" return date.strftime('%Y年%m月%d日') def load_workbook_metadata(filepath: str) -> dict: """Return workbook metadata: sheet names, date range.""" xl = pd.ExcelFile(filepath) sheets = xl.sheet_names dates = [] for s in sheets: try: d = datetime.strptime(s, '%Y年%m月%d日') dates.append(d) except ValueError: continue dates.sort() return { 'sheets': sheets, 'date_range': (dates[0], dates[-1]) if dates else (None, None), 'total_days': len(dates), } def load_daily(filepath: str, date: datetime) -> pd.DataFrame: """Load single-day order data.""" sheet = _sheet_name_for_date(date) df = pd.read_excel(filepath, sheet_name=sheet) return _clean_dataframe(df) def load_date_range(filepath: str, start: datetime, end: datetime) -> pd.DataFrame: """Load and concatenate data across a date range [start, end].""" xl = pd.ExcelFile(filepath) frames = [] current = start while current <= end: sheet = _sheet_name_for_date(current) if sheet in xl.sheet_names: df = pd.read_excel(filepath, sheet_name=sheet) df['_data_date'] = current frames.append(df) current += timedelta(days=1) if not frames: raise ValueError(f"No data found between {start.date()} and {end.date()}") combined = pd.concat(frames, ignore_index=True) return _clean_dataframe(combined) def load_weekly(filepath: str, year: int, week_num: int, week_start_day=0) -> tuple: """ Load data for a specific week. Returns (current_week_df, prev_week_df). week_start_day: 0=Monday, 6=Sunday """ # Find the first day of the given week # Simplified: assume data starts from a known reference meta = load_workbook_metadata(filepath) first_date, last_date = meta['date_range'] if first_date is None: raise ValueError("No valid date sheets found") # Find the Monday of the target week (using ISO week definition) # Jan 4 is always in week 1 jan4 = datetime(year, 1, 4) # Adjust to Monday jan4_monday = jan4 - timedelta(days=jan4.weekday()) target_monday = jan4_monday + timedelta(weeks=week_num - 1) target_sunday = target_monday + timedelta(days=6) # Clamp to available data range start = max(target_monday, first_date) end = min(target_sunday, last_date) current = load_date_range(filepath, start, end) # Previous week prev_start = start - timedelta(days=7) prev_end = end - timedelta(days=7) if prev_start >= first_date: previous = load_date_range(filepath, prev_start, prev_end) else: previous = pd.DataFrame(columns=current.columns) return current, previous def load_monthly(filepath: str, year: int, month: int) -> tuple: """ Load data for a specific month. Returns (current_month_df, prev_month_df, yoy_month_df). """ start = datetime(year, month, 1) # Last day of month if month == 12: end = datetime(year + 1, 1, 1) - timedelta(days=1) else: end = datetime(year, month + 1, 1) - timedelta(days=1) current = load_date_range(filepath, start, end) # Previous month if month == 1: prev_start = datetime(year - 1, 12, 1) prev_end = datetime(year, 1, 1) - timedelta(days=1) else: prev_start = datetime(year, month - 1, 1) prev_end = datetime(year, month, 1) - timedelta(days=1) try: previous = load_date_range(filepath, prev_start, prev_end) except ValueError: previous = pd.DataFrame(columns=current.columns) # YoY (same month last year) yoy_start = datetime(year - 1, month, 1) if month == 12: yoy_end = datetime(year, 1, 1) - timedelta(days=1) else: yoy_end = datetime(year - 1, month + 1, 1) - timedelta(days=1) try: yoy = load_date_range(filepath, yoy_start, yoy_end) except ValueError: yoy = pd.DataFrame(columns=current.columns) return current, previous, yoy def _clean_dataframe(df: pd.DataFrame) -> pd.DataFrame: """Rename columns, parse dates, clean statuses.""" # Rename known columns rename_map = {k: v for k, v in FIELD_MAP.items() if k in df.columns} df = df.rename(columns=rename_map) # Normalize status if 'status' in df.columns: df['status_code'] = df['status'].apply(_normalize_status) # Parse numeric fields if 'order_qty' in df.columns: df['order_qty'] = pd.to_numeric(df['order_qty'], errors='coerce') # Parse date fields date_fields = ['contract_date', 'deposit_date', 'order_gen_date', 'price_review_date', 'contract_review_date', 'seal_apply_date', 'seal_date', 'final_pay_date', 'customs_date', 'license_date', 'ship_date', 'invoice_date'] for field in date_fields: if field in df.columns: df[field] = df[field].apply(_parse_date) # Tracking days if 'tracking_days' in df.columns: df['tracking_days'] = pd.to_numeric(df['tracking_days'], errors='coerce') # Boolean updated if 'is_updated' in df.columns: df['is_updated_flag'] = df['is_updated'].astype(str).str.strip() == '是' return df if __name__ == '__main__': import sys if len(sys.argv) > 1: fp = sys.argv[1] meta = load_workbook_metadata(fp) print(f"Sheets: {meta['sheets'][:5]}...") print(f"Date range: {meta['date_range'][0]} ~ {meta['date_range'][1]}") print(f"Total days: {meta['total_days']}")