import uuid
from datetime import date, timedelta, datetime as dt
from typing import Optional
from sqlalchemy import text
from sqlalchemy.orm import Session


TABLE = "employment_info"
COLS = ["emp_id", "student_id", "class_id", "education", "code",
        "job_title", "company_name", "salary", "offer_date",
        "resume_open_date", "employment_int", "is_deleted",
        "create_time", "update_time"]
SELECT_COLS = ", ".join(COLS)


def _convert(d: dict) -> dict:
    if d is None:
        return None
    out = {}
    for col, val in d.items():
        if val is None:
            out[col] = None
        elif isinstance(val, dt):
            out[col] = val.isoformat()
        elif isinstance(val, date):
            out[col] = val.isoformat()
        elif hasattr(val, '__float__') and not isinstance(val, bool):
            try:
                out[col] = float(val)
            except:
                out[col] = val
        else:
            out[col] = val
    return out


def _row_to_dict(row) -> dict:
    return _convert(dict(row._mapping)) if row else None


def _rows_to_list(rows) -> list[dict]:
    return [_row_to_dict(r) for r in rows]


# ==================== 基础 CRUD ====================

def get_by_emp_id(db: Session, emp_id: str) -> Optional[dict]:
    sql = text(f"SELECT {SELECT_COLS} FROM {TABLE} WHERE emp_id = :emp_id AND is_deleted = 0")
    return _row_to_dict(db.execute(sql, {"emp_id": emp_id}).fetchone())

def list_by_student(db: Session, student_id: str) -> list[dict]:
    sql = text(f"SELECT {SELECT_COLS} FROM {TABLE} WHERE student_id = :sid AND is_deleted = 0 ORDER BY offer_date DESC")
    return _rows_to_list(db.execute(sql, {"sid": student_id}).fetchall())

def list_by_class(db: Session, class_id: str) -> list[dict]:
    sql = text(f"SELECT {SELECT_COLS} FROM {TABLE} WHERE class_id = :cid AND is_deleted = 0 ORDER BY offer_date DESC")
    return _rows_to_list(db.execute(sql, {"cid": class_id}).fetchall())

def list_employment(db: Session, filters: dict) -> list[dict]:
    conds = ["is_deleted = 0"]
    params = {}
    if filters.get("student_id"):
        conds.append("student_id = :sid"); params["sid"] = filters["student_id"]
    if filters.get("company_name"):
        conds.append("company_name LIKE :cn"); params["cn"] = f"%{filters['company_name']}%"
    if filters.get("salary_min") is not None:
        conds.append("salary >= :smin"); params["smin"] = filters["salary_min"]
    if filters.get("salary_max") is not None:
        conds.append("salary <= :smax"); params["smax"] = filters["salary_max"]
    sql = text(f"SELECT {SELECT_COLS} FROM {TABLE} WHERE {' AND '.join(conds)} ORDER BY offer_date DESC")
    return _rows_to_list(db.execute(sql, params).fetchall())

def create_by_student(db: Session, student_id: str, data_dict: dict) -> dict:
    emp_id = f"EMP{student_id}_{uuid.uuid4().hex[:8].upper()}"
    data_dict.pop("student_id", None)
    all_cols = ["emp_id", "student_id", "is_deleted"] + list(data_dict.keys())
    all_vals = [emp_id, student_id, 0] + list(data_dict.values())
    placeholders = ", ".join(f":{c}" for c in all_cols)
    col_names = ", ".join(all_cols)
    sql = text(f"INSERT INTO {TABLE} ({col_names}) VALUES ({placeholders})")
    db.execute(sql, dict(zip(all_cols, all_vals)))
    db.commit()
    return get_by_emp_id(db, emp_id)

def update_by_emp_id(db: Session, emp_id: str, updates: dict) -> Optional[dict]:
    sql = text(f"UPDATE {TABLE} SET is_deleted = 0 WHERE emp_id = :emp_id")
    db.execute(sql, {"emp_id": emp_id})
    db.commit()
    if not get_by_emp_id(db, emp_id):
        return None
    set_clause = ", ".join(f"{k} = :{k}" for k in updates.keys())
    sql = text(f"UPDATE {TABLE} SET {set_clause} WHERE emp_id = :emp_id")
    db.execute(sql, {"emp_id": emp_id, **updates})
    db.commit()
    return get_by_emp_id(db, emp_id)

def delete_by_emp_id(db: Session, emp_id: str) -> bool:
    sql = text(f"UPDATE {TABLE} SET is_deleted = 0 WHERE emp_id = :emp_id")
    db.execute(sql, {"emp_id": emp_id})
    db.commit()
    if not get_by_emp_id(db, emp_id):
        return False
    db.execute(text(f"UPDATE {TABLE} SET is_deleted = 1 WHERE emp_id = :emp_id"), {"emp_id": emp_id})
    db.commit()
    return True


# ==================== 新增功能点 ====================

def get_latest_by_student(db: Session, student_id: str) -> Optional[dict]:
    sql = text(f"SELECT {SELECT_COLS} FROM {TABLE} WHERE student_id = :sid AND is_deleted = 0 ORDER BY offer_date DESC LIMIT 1")
    return _row_to_dict(db.execute(sql, {"sid": student_id}).fetchone())

def get_class_employment_stats(db: Session, class_id: str) -> dict:
    base_sql = text(f"SELECT employment_int, COUNT(*) as cnt FROM {TABLE} WHERE class_id = :cid AND is_deleted = 0 GROUP BY employment_int")
    salary_sql = text(f"SELECT AVG(salary), MAX(salary), MIN(salary), COUNT(*) FROM {TABLE} WHERE class_id = :cid AND is_deleted = 0")
    job_sql = text(f"SELECT job_title, COUNT(*) FROM {TABLE} WHERE class_id = :cid AND is_deleted = 0 GROUP BY job_title")
    rows = db.execute(base_sql, {"cid": class_id}).fetchall()
    salary_row = db.execute(salary_sql, {"cid": class_id}).fetchone()
    job_rows = db.execute(job_sql, {"cid": class_id}).fetchall()
    stats = {r[0]: r[1] for r in rows}
    total = sum(stats.values())
    return {
        "class_id": class_id,
        "total_count": total,
        "employed_count": stats.get(1, 0),
        "graduate_count": stats.get(2, 0),
        "flexible_count": stats.get(3, 0),
        "employment_rate": round((stats.get(1, 0) + stats.get(2, 0) + stats.get(3, 0)) / total * 100, 2) if total > 0 else 0.0,
        "avg_salary": salary_row[0], "max_salary": salary_row[1], "min_salary": salary_row[2],
        "job_title_distribution": {r[0]: r[1] for r in job_rows},
    }

def get_employment_warnings(db: Session, days: int = 30, class_id: str | None = None) -> list[dict]:
    cutoff = date.today() - timedelta(days=days)
    conds = ["is_deleted = 0", "resume_open_date IS NOT NULL", "resume_open_date < :cutoff", "salary IS NULL"]
    params = {"cutoff": cutoff}
    if class_id:
        conds.append("class_id = :cid"); params["cid"] = class_id
    sql = text(f"SELECT {SELECT_COLS} FROM {TABLE} WHERE {' AND '.join(conds)}")
    result = []
    for row in db.execute(sql, params).fetchall():
        emp = _row_to_dict(row)
        overdue = (date.today() - date.fromisoformat(emp["resume_open_date"])).days - days
        result.append({
            "emp_id": emp["emp_id"], "student_id": emp["student_id"], "class_id": emp["class_id"],
            "job_title": emp["job_title"], "company_name": emp["company_name"],
            "resume_open_date": emp["resume_open_date"],
            "warning_reason": f"简历开放{days}天内未签约，已超{overdue}天",
            "days_overdue": overdue,
        })
    return result

def compare_salary_with_market(db: Session, student_id: str) -> Optional[dict]:
    emp = get_latest_by_student(db, student_id)
    if not emp or not emp.get("code") or not emp.get("job_title"):
        return None
    market_sql = text(f"SELECT COUNT(*), AVG(salary), MAX(salary), MIN(salary) FROM {TABLE} WHERE code = :code AND job_title = :jt AND salary IS NOT NULL AND is_deleted = 0")
    m = db.execute(market_sql, {"code": emp["code"], "jt": emp["job_title"]}).fetchone()
    deviation = None; advice = "无"
    if m[1] and emp.get("salary"):
        deviation = round((float(emp["salary"]) - float(m[1])) / float(m[1]) * 100, 2)
        if deviation >= 10: advice = "薪资高于行情10%+，非常优秀"
        elif deviation >= 0: advice = "薪资符合市场行情"
        elif deviation >= -10: advice = "薪资略低于行情，可考虑谈薪"
        else: advice = "薪资明显低于行情，建议与HR沟通"
    return {
        "code": emp["code"], "job_title": emp["job_title"],
        "sample_count": m[0], "avg_salary": m[1], "max_salary": m[2], "min_salary": m[3],
        "current_student_id": student_id, "current_salary": emp["salary"],
        "deviation_percent": deviation, "advice": advice,
    }