from pathlib import Path
from typing import Literal

from pydantic import Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        env_prefix="STOCK_",
        extra="ignore",
    )

    app_env: Literal["development", "test", "production"] = "development"
    reports_dir: Path = Path("reports")
    logs_dir: Path = Path("logs")
    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "INFO"
    openai_api_key: SecretStr | None = Field(
        default=None, validation_alias="OPENAI_API_KEY"
    )
    openai_model: str | None = Field(default=None, validation_alias="OPENAI_MODEL")

    def prepare_directories(self) -> None:
        self.reports_dir.mkdir(parents=True, exist_ok=True)
        self.logs_dir.mkdir(parents=True, exist_ok=True)

    def require_model_credentials(self) -> tuple[str, str]:
        key = self.openai_api_key.get_secret_value() if self.openai_api_key else ""
        model = (self.openai_model or "").strip()
        if not key.strip() or not model:
            raise ValueError("真实模型调用需要 OPENAI_API_KEY 和 OPENAI_MODEL")
        return key, model

import re


def normalize_symbol(value: str) -> str:
    symbol = value.strip().upper()
    if not re.fullmatch(r"[A-Z0-9][A-Z0-9.-]{0,19}", symbol):
        raise ValueError("股票代码需为 1 至 20 位字母、数字、点或连字符")
    return symbol

import logging
from logging.handlers import RotatingFileHandler

from stock_assistant.settings import Settings


def configure_logging(settings: Settings) -> None:
    settings.prepare_directories()
    logger = logging.getLogger("stock_assistant")
    logger.setLevel(settings.log_level)
    logger.propagate = False

    for handler in logger.handlers[:]:
        logger.removeHandler(handler)
        handler.close()

    formatter = logging.Formatter(
        "%(asctime)s %(levelname)s %(name)s %(message)s"
    )
    console = logging.StreamHandler()
    file_handler = RotatingFileHandler(
        settings.logs_dir / "app.log",
        maxBytes=5_000_000,
        backupCount=3,
        encoding="utf-8",
    )
    for handler in (console, file_handler):
        handler.setFormatter(formatter)
        logger.addHandler(handler)

import logging
from datetime import datetime, timezone
from importlib.resources import files
from pathlib import Path
from uuid import uuid4

from stock_assistant.settings import Settings
from stock_assistant.tools.symbols import normalize_symbol

logger = logging.getLogger(__name__)


def create_demo_report(symbol: str, settings: Settings) -> Path:
    normalized = normalize_symbol(symbol)
    settings.prepare_directories()
    prompt = (
        files("stock_assistant")
        .joinpath("prompts", "analyst.md")
        .read_text(encoding="utf-8")
    )
    if not prompt.strip():
        raise ValueError("分析提示词不能为空")

    now = datetime.now(timezone.utc)
    report_id = uuid4().hex
    path = settings.reports_dir / f"{normalized}-{report_id}.md"
    content = (
        f"# {normalized} 工程联通报告\n\n"
        f"- 生成时间：{now.isoformat()}\n"
        "- 运行模式：演示，无模型调用\n"
        "- 提示词资源：已成功读取\n\n"
        "## 当前结论\n\n"
        "项目已打通输入校验、配置加载、包内资源读取与 UTF-8 报告写入。\n"
        "本报告未获取实时行情、财务数据或新闻，不构成投资建议。\n\n"
        "## 待接入能力\n\n"
        "- DeepAgents 分析流程\n"
        "- 行情与财报工具\n"
        "- 证据引用及分析质量检查\n"
    )
    path.write_text(content, encoding="utf-8")
    logger.info("demo_report_created symbol=%s report_id=%s", normalized, report_id)
    return path

import argparse

from stock_assistant.logging_config import configure_logging
from stock_assistant.service import create_demo_report
from stock_assistant.settings import Settings


def main() -> None:
    parser = argparse.ArgumentParser(description="股票分析项目演示入口")
    parser.add_argument("symbol", help="例如 AAPL 或 600519.SH")
    args = parser.parse_args()

    settings = Settings()
    configure_logging(settings)
    try:
        path = create_demo_report(args.symbol, settings)
    except ValueError as exc:
        parser.error(str(exc))
    print(f"报告已生成：{path.resolve()}")


if __name__ == "__main__":
    main()

from contextlib import asynccontextmanager
from typing import Literal

from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel, Field

from stock_assistant.logging_config import configure_logging
from stock_assistant.service import create_demo_report
from stock_assistant.settings import Settings


class AnalysisRequest(BaseModel):
    symbol: str = Field(min_length=1, max_length=20)


class AnalysisResponse(BaseModel):
    mode: Literal["demo"] = "demo"
    report_file: str


@asynccontextmanager
async def lifespan(app: FastAPI):
    settings = Settings()
    configure_logging(settings)
    app.state.settings = settings
    yield


app = FastAPI(title="Stock Assistant", version="0.1.0", lifespan=lifespan)


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.post("/api/analyses", response_model=AnalysisResponse)
def create_analysis(payload: AnalysisRequest, request: Request) -> AnalysisResponse:
    try:
        path = create_demo_report(payload.symbol, request.app.state.settings)
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc
    return AnalysisResponse(report_file=path.name)

from stock_assistant.settings import Settings


def test_environment_overrides_dotenv(tmp_path, monkeypatch):
    env_file = tmp_path / ".env"
    env_file.write_text("STOCK_APP_ENV=development\n", encoding="utf-8")
    monkeypatch.setenv("STOCK_APP_ENV", "test")

    settings = Settings(_env_file=env_file)

    assert settings.app_env == "test"

import pytest

from stock_assistant.service import create_demo_report
from stock_assistant.settings import Settings


def test_demo_report_is_utf8_and_does_not_overwrite(tmp_path):
    settings = Settings(
        _env_file=None,
        app_env="test",
        log_level="INFO",
        reports_dir=tmp_path / "reports",
        logs_dir=tmp_path / "logs",
    )

    first = create_demo_report(" aapl ", settings)
    second = create_demo_report("AAPL", settings)

    assert first != second
    assert first.parent == tmp_path / "reports"
    assert first.name.startswith("AAPL-")
    assert "不构成投资建议" in first.read_text(encoding="utf-8")
    assert second.exists()


@pytest.mark.parametrize("symbol", ["../secret", "AAPL\nMSFT", "", "A" * 21])
def test_invalid_symbol_does_not_create_report(tmp_path, symbol):
    settings = Settings(
        _env_file=None,
        app_env="test",
        log_level="INFO",
        reports_dir=tmp_path / "reports",
        logs_dir=tmp_path / "logs",
    )

    with pytest.raises(ValueError):
        create_demo_report(symbol, settings)

    assert not settings.reports_dir.exists()