import duckdb
from pathlib import Path

DB_PATH = "nev_industry.db"
RAW_DIR = Path("./data/raw")
PROCESS_DIR = Path("./data/processed")

con = duckdb.connect(DB_PATH)

# ── Create tables: truncate and reload on each update ──
con.execute("""
CREATE TABLE IF NOT EXISTS sales_data (
 month VARCHAR,
 brand VARCHAR,
 model VARCHAR,
 sales_volume INTEGER,
 yoy_growth_pct DECIMAL(10,2),
 data_source VARCHAR
)
""")

con.execute("""
CREATE TABLE IF NOT EXISTS charging_stations (
 province VARCHAR,
 city VARCHAR,
 total_stations INTEGER,
 fast_chargers INTEGER,
 quarter VARCHAR
)
""")

con.execute("""
CREATE TABLE IF NOT EXISTS battery_prices (
 date VARCHAR,
 price_per_kwh DECIMAL(10,2),
 change_pct DECIMAL(10,2),
 source VARCHAR
)
""")

print("✅ Data table structure created")

def refresh_sales_data(csv_path):
 """Monthly sales data update"""
 con.execute("TRUNCATE TABLE sales_data")
 con.execute(f"""
 INSERT INTO sales_data
 SELECT * FROM read_csv_auto('{csv_path}', 
 headers=true, 
 autoparse=true,
 sep=','
 )
 """)
 print(f"✅ Sales data updated: {con.execute('SELECT COUNT(*) FROM sales_data').fetchone()[0]} records")

def refresh_charging_data(json_path):
 """Update charging station data"""
 con.execute("TRUNCATE TABLE charging_stations")
 con.execute(f"""
 INSERT INTO charging_stations
 SELECT * FROM read_json_auto('{json_path}')
 """)
 print(f"✅ Charging station data updated")

def refresh_battery_prices(csv_path):
 """Update battery price data"""
 con.execute("TRUNCATE TABLE battery_prices")
 con.execute(f"""
 INSERT INTO battery_prices
 SELECT * FROM read_csv_auto('{csv_path}', 
 headers=true, 
 timezone='Asia/Shanghai'
 )
 """)
 print(f"✅ Battery price data updated")

def generate_weekly_report():
 """Generate core data for weekly report"""
 
 # ── Metric 1: Top 10 Brands by Sales ──
 top_brands = con.execute("""
 SELECT 
 brand,
 SUM(sales_volume) AS total_sales,
 ROUND(AVG(yoy_growth_pct), 1) AS avg_yoy_growth,
 COUNT(DISTINCT model) AS model_count
 FROM sales_data
 WHERE month = (SELECT MAX(month) FROM sales_data)
 GROUP BY brand
 ORDER BY total_sales DESC
 LIMIT 10
 """).fetchdf()
 
 # ── Metric 2: Price Segment Market Share ──
 price_segment = con.execute("""
 WITH order_data AS (
 SELECT 
 model,
 brand,
 CASE 
 WHEN avg_price < 10 THEN '<100k'
 WHEN avg_price < 20 THEN '100k-200k'
 WHEN avg_price < 30 THEN '200k-300k'
 ELSE '300k+'
 END AS price_segment,
 SUM(sales_volume) AS sales
 FROM (
 SELECT 
 s.model,
 s.brand,
 s.sales_volume,
 CASE 
 WHEN s.brand IN ('BYD Dolphin', 'Wuling Bingo') THEN 8
 WHEN s.brand IN ('BYD Qin PLUS', 'GAC Aion Y') THEN 12
 WHEN s.brand IN ('BYD Han', 'XPeng P7') THEN 20
 ELSE 35
 END AS avg_price
 FROM sales_data s
 WHERE s.month = (SELECT MAX(month) FROM sales_data)
 ) sub
 GROUP BY model, brand, price_segment
 )
 SELECT 
 price_segment,
 SUM(sales) AS total_sales,
 ROUND(SUM(sales) * 100.0 / NULLIF(SUM(SUM(sales)) OVER (), 0), 1) AS market_share_pct
 FROM order_data
 GROUP BY price_segment
 ORDER BY total_sales DESC
 """).fetchdf()
 
 # ── Metric 3: Charging Infrastructure Coverage ──
 charging_coverage = con.execute("""
 SELECT 
 province,
 SUM(total_stations) AS total_stations,
 SUM(fast_chargers) AS total_fast_chargers,
 ROUND(SUM(fast_chargers) * 100.0 / NULLIF(SUM(total_stations), 0), 2) AS fast_ratio_pct
 FROM charging_stations
 WHERE quarter = (SELECT MAX(quarter) FROM charging_stations)
 GROUP BY province
 ORDER BY total_stations DESC
 LIMIT 10
 """).fetchdf()
 
 # ── Metric 4: Battery Cost Trend (Key Signal for Vehicle Profit) ──
 battery_trend = con.execute("""
 SELECT 
 date,
 price_per_kwh,
 change_pct,
 ROUND(AVG(price_per_kwh) OVER (
 ORDER BY date 
 ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
 ), 2) AS ma_30d
 FROM battery_prices
 ORDER BY date DESC
 LIMIT 14
 """).fetchdf()
 
 return {
 'top_brands': top_brands,
 'price_segment': price_segment,
 'charging_coverage': charging_coverage,
 'battery_trend': battery_trend
 }

def format_report_text(data):
 """Convert data into natural language briefing"""
 import datetime
 lines = []
 lines.append("📊 New Energy Vehicle Industry Weekly | " + datetime.date.today().strftime("%Y-%m-%d"))
 lines.append("")
 
 # ── Headline: This Week's Biggest Story ──
 top_brand = data['top_brands'].iloc[0]
 lines.append(f"🏆 Spotlight: {top_brand['brand']} leads with {int(top_brand['total_sales']):,} units sold, up {top_brand['avg_yoy_growth']}% year-over-year")
 lines.append("")
 
 # ── Segment Insights ──
 lines.append("📈 Price Segment Competition:")
 for _, row in data['price_segment'].iterrows():
 emoji = "🔥" if row['market_share_pct'] > 30 else "📌"
 lines.append(f" {emoji} {row['price_segment']}: {int(row['total_sales']):,} units ({row['market_share_pct']}% share)")
 lines.append("")
 
 # ── Infrastructure Progress ──
 lines.append("🔋 Charging Infrastructure (Top 3 Provinces):")
 for _, row in data['charging_coverage'].head(3).iterrows():
 lines.append(f" {row['province']}: {int(row['total_stations']):,} stations, {row['fast_ratio_pct']}% fast chargers")
 lines.append("")
 
 # ── Cost Signals ──
 latest_battery = data['battery_trend'].iloc[0]
 prev_battery = data['battery_trend'].iloc[1] if len(data['battery_trend']) > 1 else latest_battery
 change_note = "📉 Down" if latest_battery['change_pct'] < 0 else "📈 Up"
 lines.append(f"⚡ Battery Cost Signal: Lithium carbonate at {latest_battery['price_per_kwh']} RMB/kWh ({change_note} {abs(latest_battery['change_pct'])}%), 30-day avg {latest_battery['ma_30d']} RMB/kWh")
 lines.append("")
 
 # ── Key Conclusions (Most Valuable Part) ──
 lines.append("💡 Key Takeaways:")
 
 # Auto-generated insight logic
 if top_brand['avg_yoy_growth'] > 20:
 lines.append(" 1. Top brands still maintaining high growth, industry concentration increasing")
 if len(data['price_segment']) > 0 and data['price_segment'].iloc[0]['price_segment'] == '100k-200k':
 lines.append(" 2. 100k-200k RMB segment remains the main competitive battlefield")
 if latest_battery['change_pct'] < -2:
 lines.append(" 3. Rapid battery cost decline may ease price war pressure H2")
 
 lines.append("")
 lines.append("---")
 lines.append("Data Sources: CAAM, MIIT, Brand Announcements, Industry Public Data")
 lines.append("Report auto-generated by DuckDB, Data as of latest reporting period")
 
 return "\n".join(lines)

# Generate and save the report
report_data = generate_weekly_report()
report_text = format_report_text(report_data)

# Save as Markdown
output_path = f"./reports/report_{datetime.date.today().strftime('%Y%m%d')}.md"
with open(output_path, 'w', encoding='utf-8') as f:
 f.write(report_text)

print(f"✅ Report generated: {output_path}")