#!/usr/bin/env python3
"""Test worker: pull deduced profit for a small batch of stocks.
Usage: python3 test_worker_deduced_profit.py worker_41 000333 000001 000002 000858 002594
"""
import sys
import os
import time
import csv
import re
import akshare as ak

WORKER_ID = sys.argv[1]
CODES = sys.argv[2:]
OUTPUT_DIR = os.path.expanduser(f"~/elite_stocks_import/{WORKER_ID}")
os.makedirs(OUTPUT_DIR, exist_ok=True)

LOG_FILE = os.path.join(OUTPUT_DIR, f"{WORKER_ID}.log")

def log(msg):
    print(msg)
    with open(LOG_FILE, 'a', encoding='utf-8') as f:
        f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')} {msg}\n")

def parse_amount(s):
    """Parse strings like '412.67亿', '-12.5%', 'False' to float."""
    if s is None or s is False or str(s).strip() == 'False':
        return None
    s = str(s).strip()
    s = s.replace(',', '')
    if s.endswith('亿'):
        return float(s[:-1])
    if s.endswith('万'):
        return float(s[:-1]) / 10000
    if s.endswith('千'):
        return float(s[:-1]) / 100000
    if s.endswith('%'):
        return float(s[:-1])
    try:
        return float(s)
    except:
        return None

def pull_stock(code):
    try:
        log(f"[{code}] pulling...")
        time.sleep(2.5)
        df = ak.stock_financial_abstract_ths(symbol=code, indicator='按单季度')
        if df is None or df.empty:
            log(f"[{code}] empty result")
            return False
        
        column_map = {
            '报告期': 'report_date',
            '净利润': 'net_profit',
            '净利润同比增长率': 'net_profit_yoy',
            '扣非净利润': 'deduced_profit',
            '扣非净利润同比增长率': 'deduced_profit_yoy',
            '营业总收入': 'total_revenue',
            '营业总收入同比增长率': 'total_revenue_yoy',
            '基本每股收益': 'basic_eps',
            '每股净资产': 'bps',
            '净资产收益率-摊薄': 'roe_diluted',
            '销售毛利率': 'gross_margin',
            '销售净利率': 'net_margin',
        }
        
        # Only select columns that exist in the returned dataframe
        available_cols = [c for c in column_map.keys() if c in df.columns]
        selected = df[available_cols].copy()
        selected.rename(columns={c: column_map[c] for c in available_cols}, inplace=True)
        
        # Add missing columns as None
        for en_col in column_map.values():
            if en_col not in selected.columns:
                selected[en_col] = None
        
        selected['code'] = code
        selected['name'] = ''
        selected['source'] = 'akshare_ths'
        selected['fetched_by'] = WORKER_ID
        selected['fetched_at'] = time.strftime('%Y-%m-%d %H:%M:%S')
        
        numeric_cols = ['net_profit', 'net_profit_yoy', 'deduced_profit', 'deduced_profit_yoy',
                        'total_revenue', 'total_revenue_yoy', 'basic_eps', 'bps', 
                        'roe_diluted', 'gross_margin', 'net_margin']
        for col in numeric_cols:
            if col in selected.columns:
                selected[col] = selected[col].apply(parse_amount)
        
        cols = ['code', 'name', 'report_date', 'net_profit', 'net_profit_yoy', 
                'deduced_profit', 'deduced_profit_yoy', 'total_revenue', 'total_revenue_yoy',
                'basic_eps', 'bps', 'roe_diluted', 'gross_margin', 'net_margin',
                'source', 'fetched_by', 'fetched_at']
        selected = selected[cols]
        
        output_file = os.path.join(OUTPUT_DIR, f"deduced_profit_{code}.csv")
        selected.to_csv(output_file, index=False, encoding='utf-8')
        log(f"[{code}] saved {len(selected)} rows to {output_file}")
        return True
    except Exception as e:
        log(f"[{code}] error: {e}")
        return False

def main():
    log(f"Worker {WORKER_ID} started with {len(CODES)} codes: {CODES}")
    success = 0
    for code in CODES:
        if pull_stock(code):
            success += 1
    log(f"Worker {WORKER_ID} finished: {success}/{len(CODES)} success")

if __name__ == '__main__':
    main()