use crate::cli::Cli;
use crate::source::{ResolvedInput, is_url, resolve_input};
use anyhow::{Context, Result, anyhow};
use glob::{MatchOptions, glob_with};
use spoor_core::{
    Format, JsonOutput, OutputMode, ParseContent, ParseLimits, ParseRequest, SpoorError,
    SpoorWarning, TableFilter, default_mode_for, detect_format, extract_media,
    limit_markdown_output, parse_document_result, parse_tables, render_documents,
    render_json_limited,
};

const MAX_FAILURE_DIAGNOSTICS: usize = 20;

pub(crate) enum CommandOutput {
    Text(String),
    Binary(Vec<u8>),
}

pub(crate) fn run(cli: Cli) -> Result<CommandOutput> {
    validate_max_parse_bytes(cli.max_parse_bytes)?;
    if let Some(resource) = cli.extract.clone() {
        return extract_resource(&cli, &resource).map(CommandOutput::Binary);
    }
    validate_max_output_bytes(cli.max_output_bytes)?;
    run_parse(cli).map(CommandOutput::Text)
}

fn extract_resource(cli: &Cli, resource: &str) -> Result<Vec<u8>> {
    let inputs = expand_inputs(&cli.inputs)?;
    if inputs.len() != 1 {
        return Err(anyhow!("--extract 仅支持单个输入"));
    }
    let input = resolve_input(&inputs[0], cli.max_parse_bytes)?;
    let request = request_for(&input, None, TableFilter::default(), cli.max_parse_bytes);
    extract_media(&request, resource).map_err(Into::into)
}

fn run_parse(cli: Cli) -> Result<String> {
    let inputs = expand_inputs(&cli.inputs)?;
    let format_hint = cli.format.map(Into::into);

    // Resolve each input independently: one unreadable file or failed fetch
    // must not abort the whole batch. Failures are collected and surfaced as
    // warnings (stderr, plus the JSON envelope in json mode) so the remaining
    // inputs still produce output. We only fail hard (exit 1) when *nothing*
    // succeeds.
    let mut resolved = Vec::with_capacity(inputs.len());
    let mut failures = Vec::new();
    let mut source_bytes = 0usize;
    for input in inputs {
        let remaining = cli.max_parse_bytes.saturating_sub(source_bytes);
        match resolve_input(&input, remaining) {
            Ok(input_data) => {
                let request =
                    request_for(&input_data, format_hint, TableFilter::default(), remaining);
                match detect_format(&request) {
                    Ok(format) => {
                        source_bytes += input_data.len();
                        resolved.push(PreparedInput {
                            input: input_data,
                            format,
                            max_parse_bytes: remaining,
                        });
                    }
                    Err(error) => failures.push(InputFailure::from_spoor(input, &error)),
                }
            }
            Err(e) => failures.push(InputFailure::from_error(input, &e)),
        }
    }

    let formats: Vec<_> = resolved.iter().map(|resolved| resolved.format).collect();
    let mode = cli
        .mode
        .map(Into::into)
        .unwrap_or_else(|| default_mode_for(&formats));

    match mode {
        OutputMode::Md => {
            warn_unused_narrowing(&cli);
            let mut documents = Vec::with_capacity(resolved.len());
            let mut parse_warnings = Vec::new();
            let mut extracted_bytes = 0usize;
            for resolved in &resolved {
                let request = request_for(
                    &resolved.input,
                    Some(resolved.format),
                    TableFilter::default(),
                    resolved.max_parse_bytes,
                );
                match parse_document_result(&request) {
                    Ok(result) => {
                        let ParseContent::Document(document) = result.content else {
                            failures.push(InputFailure::from_error(
                                resolved.input.label.clone(),
                                &anyhow!("内部错误：Markdown 模式返回了表格结果"),
                            ));
                            continue;
                        };
                        if let Err(error) = retain_within_parse_budget(
                            &mut extracted_bytes,
                            document.markdown.len(),
                            cli.max_parse_bytes,
                            "retained extracted documents",
                        ) {
                            failures.push(InputFailure::from_error(
                                resolved.input.label.clone(),
                                &error,
                            ));
                        } else {
                            parse_warnings.extend(result.warnings.into_iter().map(|warning| {
                                InputWarning {
                                    source: resolved.input.label.clone(),
                                    warning,
                                }
                            }));
                            documents.push(document);
                        }
                    }
                    Err(error) => failures.push(InputFailure::from_spoor(
                        resolved.input.label.clone(),
                        &error,
                    )),
                }
            }
            if documents.is_empty() {
                return Err(all_failed_error(&failures));
            }
            report_skipped(&failures);
            report_parse_warnings(&parse_warnings);
            let markdown = render_documents(&documents, mode)?;
            // Reserve room for in-band diagnostics so the total remains within
            // --max-output-bytes and agents do not lose warnings first.
            let diagnostics = [
                markdown_skipped_block(&failures),
                markdown_parse_warnings_block(&parse_warnings),
            ]
            .into_iter()
            .flatten()
            .collect::<String>();
            let limited_diagnostics = limit_markdown_output(diagnostics, cli.max_output_bytes);
            report_output_truncation(limited_diagnostics.warning.as_deref());
            let budget = cli
                .max_output_bytes
                .saturating_sub(limited_diagnostics.content.len());
            let limited = limit_markdown_output(markdown, budget);
            report_output_truncation(limited.warning.as_deref());
            let mut content = limited.content;
            content.push_str(&limited_diagnostics.content);
            Ok(content)
        }
        OutputMode::Json => {
            let filter = build_filter(&cli)?;
            let mut tables = Vec::new();
            let mut successes = 0usize;
            let mut extracted_bytes = 0usize;
            for resolved in &resolved {
                let request = request_for(
                    &resolved.input,
                    Some(resolved.format),
                    filter.clone(),
                    resolved.max_parse_bytes,
                );
                match parse_tables(&request) {
                    Ok(extracted) => {
                        if let Err(error) = retain_within_parse_budget(
                            &mut extracted_bytes,
                            extracted.serialized_bytes,
                            cli.max_parse_bytes,
                            "retained extracted tables",
                        ) {
                            failures.push(InputFailure::from_error(
                                resolved.input.label.clone(),
                                &error,
                            ));
                        } else {
                            successes += 1;
                            tables.extend(extracted.tables);
                        }
                    }
                    Err(error) => failures.push(InputFailure::from_spoor(
                        resolved.input.label.clone(),
                        &error,
                    )),
                }
            }
            if successes == 0 {
                return Err(all_failed_error(&failures));
            }
            report_skipped(&failures);
            let mut output = JsonOutput::new(tables);
            output.warnings = failures.iter().map(InputFailure::display).collect();
            let limited = render_json_limited(&output, cli.max_output_bytes);
            report_output_truncation(limited.warning.as_deref());
            Ok(limited.content)
        }
    }
}

#[derive(Debug)]
struct InputWarning {
    source: String,
    warning: SpoorWarning,
}

impl InputWarning {
    fn display(&self) -> String {
        let warning = serde_json::to_string(&self.warning)
            .unwrap_or_else(|_| format!("{}: {}", self.warning.code, self.warning.message));
        format!("{}: {warning}", self.source)
    }
}

fn validate_max_output_bytes(max_output_bytes: usize) -> Result<()> {
    if max_output_bytes < spoor_core::MIN_MAX_OUTPUT_BYTES {
        return Err(anyhow!(
            "--max-output-bytes 不能小于 {}",
            spoor_core::MIN_MAX_OUTPUT_BYTES
        ));
    }
    Ok(())
}

fn validate_max_parse_bytes(max_parse_bytes: usize) -> Result<()> {
    if max_parse_bytes < spoor_core::MIN_MAX_PARSE_BYTES {
        return Err(anyhow!(
            "--max-parse-bytes 不能小于 {}",
            spoor_core::MIN_MAX_PARSE_BYTES
        ));
    }
    Ok(())
}

fn retain_within_parse_budget(
    retained: &mut usize,
    additional: usize,
    max_parse_bytes: usize,
    stage: &str,
) -> Result<()> {
    let total = retained.checked_add(additional).unwrap_or(usize::MAX);
    if total > max_parse_bytes {
        return Err(SpoorError::parse_memory_limit(max_parse_bytes, stage).into());
    }
    *retained = total;
    Ok(())
}

fn report_output_truncation(warning: Option<&str>) {
    if let Some(warning) = warning {
        eprintln!("warning: {warning}");
    }
}

#[derive(Debug)]
struct InputFailure {
    source: String,
    message: String,
    structured: Option<SpoorError>,
}

impl InputFailure {
    fn from_error(source: String, error: &anyhow::Error) -> Self {
        let structured = error.downcast_ref::<SpoorError>().cloned();
        let message = structured
            .as_ref()
            .map(SpoorError::to_json)
            .unwrap_or_else(|| error.root_cause().to_string());

        Self {
            source,
            message,
            structured,
        }
    }

    fn from_spoor(source: String, error: &SpoorError) -> Self {
        Self {
            source,
            message: error.to_json(),
            structured: Some(error.clone()),
        }
    }

    fn display(&self) -> String {
        format!("{}: {}", self.source, self.message)
    }
}

/// Emit one stderr line per skipped input. In json mode the same messages
/// also land in the envelope's top-level `warnings[]`; stderr keeps a human
/// running the command in a terminal informed regardless of output mode.
fn report_skipped(failures: &[InputFailure]) {
    for failure in failures.iter().take(MAX_FAILURE_DIAGNOSTICS) {
        eprintln!("warning: 已跳过 {}", failure.display());
    }
    if failures.len() > MAX_FAILURE_DIAGNOSTICS {
        eprintln!(
            "warning: 另有 {} 条被跳过输入的警告未显示",
            failures.len() - MAX_FAILURE_DIAGNOSTICS
        );
    }
}

fn report_parse_warnings(warnings: &[InputWarning]) {
    for warning in warnings.iter().take(MAX_FAILURE_DIAGNOSTICS) {
        eprintln!("warning: 解析结果不完整 {}", warning.display());
    }
    if warnings.len() > MAX_FAILURE_DIAGNOSTICS {
        eprintln!(
            "warning: 另有 {} 条解析完整性警告未显示",
            warnings.len() - MAX_FAILURE_DIAGNOSTICS
        );
    }
}

/// In-band counterpart of `report_skipped` for markdown mode. Agents often
/// read only stdout (e.g. redirected to a file), so a partial-batch failure
/// must be visible there too — mirroring the JSON envelope's `warnings[]`.
fn markdown_skipped_block(failures: &[InputFailure]) -> Option<String> {
    if failures.is_empty() {
        return None;
    }

    let mut block = format!(
        "\n> [!WARNING]\n> spoor 已跳过 {} 个无法读取的输入：\n",
        failures.len()
    );
    for failure in failures.iter().take(MAX_FAILURE_DIAGNOSTICS) {
        block.push_str(&format!(
            "> - {}\n",
            failure.display().replace(['\r', '\n'], " ")
        ));
    }
    if failures.len() > MAX_FAILURE_DIAGNOSTICS {
        block.push_str(&format!(
            "> - …… 另有 {} 个失败未列出\n",
            failures.len() - MAX_FAILURE_DIAGNOSTICS
        ));
    }

    Some(block)
}

fn markdown_parse_warnings_block(warnings: &[InputWarning]) -> Option<String> {
    if warnings.is_empty() {
        return None;
    }

    let mut block = format!(
        "\n> [!WARNING]\n> spoor 返回 {} 条解析完整性警告；Agent 不应把受影响位置视为完整原文：\n",
        warnings.len()
    );
    for warning in warnings.iter().take(MAX_FAILURE_DIAGNOSTICS) {
        block.push_str(&format!(
            "> - {}\n",
            warning.display().replace(['\r', '\n'], " ")
        ));
    }
    if warnings.len() > MAX_FAILURE_DIAGNOSTICS {
        block.push_str(&format!(
            "> - …… 另有 {} 条解析完整性警告未列出\n",
            warnings.len() - MAX_FAILURE_DIAGNOSTICS
        ));
    }

    Some(block)
}

/// Error returned when every input failed (exit 1). A lone failure is
/// surfaced verbatim; multiple are aggregated into one message.
fn all_failed_error(failures: &[InputFailure]) -> anyhow::Error {
    match failures {
        [
            InputFailure {
                structured: Some(error),
                ..
            },
        ] => anyhow::Error::new(error.clone()),
        [only] => anyhow!(only.display()),
        _ => anyhow!(
            "全部 {} 个输入均失败：\n  - {}{}",
            failures.len(),
            failures
                .iter()
                .take(MAX_FAILURE_DIAGNOSTICS)
                .map(InputFailure::display)
                .collect::<Vec<_>>()
                .join("\n  - "),
            if failures.len() > MAX_FAILURE_DIAGNOSTICS {
                format!(
                    "\n  - …… 另有 {} 个失败未显示",
                    failures.len() - MAX_FAILURE_DIAGNOSTICS
                )
            } else {
                String::new()
            }
        ),
    }
}

fn build_filter(cli: &Cli) -> Result<TableFilter> {
    let row_range = match &cli.rows {
        Some(s) => Some(parse_row_range(s)?),
        None => None,
    };

    Ok(TableFilter {
        sheet: cli.sheet.clone(),
        row_range,
        columns: cli.columns.clone(),
        limit: cli.limit,
        offset: cli.offset,
    })
}

fn parse_row_range(s: &str) -> Result<(usize, usize)> {
    let (first, last) = s
        .split_once(':')
        .ok_or_else(|| anyhow!("--rows expects <first>:<last>, got {s:?}"))?;
    let first: usize = first
        .trim()
        .parse()
        .map_err(|_| anyhow!("--rows: invalid first row {first:?}"))?;
    let last: usize = last
        .trim()
        .parse()
        .map_err(|_| anyhow!("--rows: invalid last row {last:?}"))?;
    if first == 0 || last == 0 {
        return Err(anyhow!("--rows: row numbers must be >= 1, got {s:?}"));
    }
    if first > last {
        return Err(anyhow!("--rows: first ({first}) > last ({last})"));
    }
    Ok((first, last))
}

fn warn_unused_narrowing(cli: &Cli) {
    let used = cli.sheet.is_some()
        || cli.rows.is_some()
        || !cli.columns.is_empty()
        || cli.limit.is_some()
        || cli.offset.is_some();
    if used {
        eprintln!("warning: Markdown 模式下会忽略 --sheet/--rows/--columns/--limit/--offset");
    }
}

fn expand_inputs(inputs: &[String]) -> Result<Vec<String>> {
    let mut expanded = Vec::new();

    for input in inputs {
        if is_url(input) || !has_glob_meta(input) {
            expanded.push(input.clone());
            continue;
        }

        expanded
            .extend(expand_glob(input).with_context(|| format!("failed to expand glob: {input}"))?);
    }

    Ok(expanded)
}

fn expand_glob(pattern: &str) -> Result<Vec<String>> {
    let options = MatchOptions {
        case_sensitive: !cfg!(windows),
        require_literal_separator: true,
        require_literal_leading_dot: false,
    };

    let mut matches = glob_with(pattern, options)
        .with_context(|| format!("invalid glob pattern: {pattern}"))?
        .collect::<std::result::Result<Vec<_>, _>>()
        .with_context(|| format!("failed to read glob matches: {pattern}"))?
        .into_iter()
        .map(|path| path.to_string_lossy().into_owned())
        .collect::<Vec<_>>();

    matches.sort();
    matches.dedup();

    if matches.is_empty() {
        return Err(anyhow!("glob matched no files: {pattern}"));
    }

    Ok(matches)
}

fn has_glob_meta(s: &str) -> bool {
    s.contains('*') || s.contains('?') || s.contains('[')
}

struct PreparedInput {
    input: ResolvedInput,
    format: Format,
    max_parse_bytes: usize,
}

fn request_for<'a>(
    input: &'a ResolvedInput,
    format_hint: Option<Format>,
    table_filter: TableFilter,
    max_parse_bytes: usize,
) -> ParseRequest<'a> {
    ParseRequest {
        bytes: &input.bytes,
        source_name: Some(&input.label),
        content_type: input.content_type.as_deref(),
        format_hint,
        table_filter,
        limits: ParseLimits { max_parse_bytes },
    }
}

#[cfg(test)]
struct TestDir {
    path: std::path::PathBuf,
}

#[cfg(test)]
impl TestDir {
    fn new(name: &str) -> Result<Self> {
        let unique = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)?
            .as_nanos();
        let path = std::env::temp_dir().join(format!("spoor-{name}-{unique}"));
        std::fs::create_dir_all(&path)?;
        Ok(Self { path })
    }

    fn path(&self) -> &std::path::Path {
        &self.path
    }
}

#[cfg(test)]
impl Drop for TestDir {
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.path);
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn glob_inputs_expand_and_sort() -> Result<()> {
        let dir = TestDir::new("glob_inputs_expand_and_sort")?;
        std::fs::write(dir.path().join("b.pdf"), b"")?;
        std::fs::write(dir.path().join("a.pdf"), b"")?;
        std::fs::write(dir.path().join("notes.txt"), b"")?;

        let pattern = dir.path().join("*.pdf").to_string_lossy().into_owned();
        let expanded = expand_inputs(&[pattern])?;
        let names = expanded
            .iter()
            .map(|path| {
                std::path::Path::new(path)
                    .file_name()
                    .unwrap()
                    .to_string_lossy()
                    .into_owned()
            })
            .collect::<Vec<_>>();

        assert_eq!(names, ["a.pdf", "b.pdf"]);
        Ok(())
    }

    #[test]
    fn glob_without_matches_errors() -> Result<()> {
        let dir = TestDir::new("glob_without_matches_errors")?;
        let pattern = dir.path().join("*.pdf").to_string_lossy().into_owned();
        let err = expand_inputs(&[pattern]).unwrap_err();

        assert!(err.to_string().contains("failed to expand glob"));
        assert!(format!("{err:#}").contains("glob matched no files"));
        Ok(())
    }

    #[test]
    fn parse_row_range_accepts_valid_input() {
        assert_eq!(parse_row_range("5:104").unwrap(), (5, 104));
        assert_eq!(parse_row_range("1:1").unwrap(), (1, 1));
        assert_eq!(parse_row_range(" 5 : 104 ").unwrap(), (5, 104));
    }

    #[test]
    fn parse_row_range_rejects_invalid_input() {
        assert!(parse_row_range("5").is_err());
        assert!(parse_row_range("a:b").is_err());
        assert!(parse_row_range("104:5").is_err());
        assert!(parse_row_range("0:10").is_err());
    }

    #[test]
    fn all_failed_error_caps_diagnostics() {
        let failures = (0..25)
            .map(|index| InputFailure {
                source: format!("input-{index}"),
                message: "failed".to_string(),
                structured: None,
            })
            .collect::<Vec<_>>();

        let message = all_failed_error(&failures).to_string();
        assert!(message.contains("全部 25 个输入均失败"));
        assert!(message.contains("input-19"));
        assert!(!message.contains("input-20"));
        assert!(message.contains("另有 5 个失败未显示"));
    }

    #[test]
    fn retained_results_share_parse_budget() {
        let mut retained = 600;
        let error = retain_within_parse_budget(&mut retained, 500, 1024, "test").unwrap_err();

        assert_eq!(retained, 600);
        assert!(error.downcast_ref::<SpoorError>().is_some());
    }
}
