use std::path::Path;
use std::sync::Arc;

use reqwest::Url;
use serde_json::{Map, Value};
use zeroclaw::skills::{load_skills_from_directory, SkillTool};

use crate::browser::{BrowserBackend, PipeBrowserBackend};
use crate::compat::browser_script_skill_tool::execute_browser_script_tool;
use crate::compat::config_adapter::resolve_skills_dir_from_sgclaw_settings;
use crate::compat::runtime::CompatTaskContext;
use crate::config::SgClawSettings;
use crate::pipe::{BrowserPipeTool, PipeError, Transport};

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DirectSubmitOutcome {
    pub success: bool,
    pub summary: String,
}

pub fn execute_direct_submit_skill<T: Transport + 'static>(
    browser_tool: BrowserPipeTool<T>,
    instruction: &str,
    task_context: &CompatTaskContext,
    workspace_root: &Path,
    settings: &SgClawSettings,
) -> Result<DirectSubmitOutcome, PipeError> {
    let configured_tool = settings
        .direct_submit_skill
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .ok_or_else(|| PipeError::Protocol("direct submit skill is not configured".to_string()))?;
    let expected_domain = derive_expected_domain(task_context)?;
    let period = derive_period(instruction)?;

    let mut args = Map::new();
    args.insert("expected_domain".to_string(), Value::String(expected_domain));
    args.insert("period".to_string(), Value::String(period));

    let output = execute_browser_script_skill_raw_output(
        browser_tool,
        configured_tool,
        workspace_root,
        settings,
        args,
    )?;

    Ok(interpret_direct_submit_output(&output))
}

pub fn execute_direct_submit_skill_with_browser_backend(
    browser_backend: Arc<dyn BrowserBackend>,
    instruction: &str,
    task_context: &CompatTaskContext,
    workspace_root: &Path,
    settings: &SgClawSettings,
) -> Result<DirectSubmitOutcome, PipeError> {
    let configured_tool = settings
        .direct_submit_skill
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .ok_or_else(|| PipeError::Protocol("direct submit skill is not configured".to_string()))?;
    let expected_domain = derive_expected_domain(task_context)?;
    let period = derive_period(instruction)?;

    let mut args = Map::new();
    args.insert("expected_domain".to_string(), Value::String(expected_domain));
    args.insert("period".to_string(), Value::String(period));

    let output = execute_browser_script_skill_raw_output_with_browser_backend(
        browser_backend,
        configured_tool,
        workspace_root,
        settings,
        args,
    )?;

    Ok(interpret_direct_submit_output(&output))
}

pub fn execute_browser_script_skill_raw_output<T: Transport + 'static>(
    browser_tool: BrowserPipeTool<T>,
    configured_tool: &str,
    workspace_root: &Path,
    settings: &SgClawSettings,
    args: Map<String, Value>,
) -> Result<String, PipeError> {
    let (tool, skill_root) = resolve_browser_script_skill(configured_tool, workspace_root, settings)?;

    execute_browser_script_tool_output(browser_tool, configured_tool, &tool, &skill_root, args)
}

pub fn execute_browser_script_skill_raw_output_with_browser_backend(
    browser_backend: Arc<dyn BrowserBackend>,
    configured_tool: &str,
    workspace_root: &Path,
    settings: &SgClawSettings,
    args: Map<String, Value>,
) -> Result<String, PipeError> {
    let (tool, skill_root) =
        resolve_browser_script_skill(configured_tool, workspace_root, settings)?;

    execute_browser_script_tool_output_with_backend(
        browser_backend.as_ref(),
        configured_tool,
        &tool,
        &skill_root,
        args,
    )
}

fn resolve_browser_script_skill(
    configured_tool: &str,
    workspace_root: &Path,
    settings: &SgClawSettings,
) -> Result<(SkillTool, std::path::PathBuf), PipeError> {
    let (skill_name, tool_name) = parse_configured_tool_name(configured_tool)?;
    let skills_dir = resolve_skills_dir_from_sgclaw_settings(workspace_root, settings);
    let skills = load_skills_from_directory(&skills_dir, true);
    let skill = skills
        .into_iter()
        .find(|skill| skill.name == skill_name)
        .ok_or_else(|| {
            PipeError::Protocol(format!(
                "direct submit skill {skill_name} was not found in {}",
                skills_dir.display()
            ))
        })?;
    let skill_root = skill
        .location
        .as_deref()
        .and_then(Path::parent)
        .map(Path::to_path_buf)
        .ok_or_else(|| {
            PipeError::Protocol(format!(
                "direct submit skill {skill_name} is missing a resolvable location"
            ))
        })?;
    let tool = skill
        .tools
        .iter()
        .find(|tool| tool.name == tool_name)
        .cloned()
        .ok_or_else(|| {
            PipeError::Protocol(format!(
                "direct submit tool {configured_tool} was not found"
            ))
        })?;

    Ok((tool, skill_root))
}

fn execute_browser_script_tool_output<T: Transport + 'static>(
    browser_tool: BrowserPipeTool<T>,
    configured_tool: &str,
    tool: &SkillTool,
    skill_root: &Path,
    args: Map<String, Value>,
) -> Result<String, PipeError> {
    let browser_backend = PipeBrowserBackend::from_inner(browser_tool);
    execute_browser_script_tool_output_with_backend(
        &browser_backend,
        configured_tool,
        tool,
        skill_root,
        args,
    )
}

fn execute_browser_script_tool_output_with_backend(
    browser_backend: &dyn BrowserBackend,
    configured_tool: &str,
    tool: &SkillTool,
    skill_root: &Path,
    args: Map<String, Value>,
) -> Result<String, PipeError> {
    if tool.kind != "browser_script" {
        return Err(PipeError::Protocol(format!(
            "direct submit tool {configured_tool} must be browser_script, got {}",
            tool.kind
        )));
    }

    let mut tool = tool.clone();
    tool.args.remove("expected_domain");

    let runtime = tokio::runtime::Runtime::new()
        .map_err(|err| PipeError::Protocol(format!("failed to create tokio runtime: {err}")))?;
    let result = runtime
        .block_on(execute_browser_script_tool(
            &tool,
            skill_root,
            browser_backend,
            Value::Object(args),
        ))
        .map_err(|err| PipeError::Protocol(err.to_string()))?;

    if result.success {
        Ok(result.output)
    } else {
        Err(PipeError::Protocol(
            result
                .error
                .unwrap_or_else(|| "direct submit skill execution failed".to_string()),
        ))
    }
}

fn interpret_direct_submit_output(output: &str) -> DirectSubmitOutcome {
    let Some(payload) = serde_json::from_str::<Value>(output).ok() else {
        return DirectSubmitOutcome {
            success: true,
            summary: output.to_string(),
        };
    };

    let Some(artifact) = payload.as_object() else {
        return DirectSubmitOutcome {
            success: true,
            summary: output.to_string(),
        };
    };

    if artifact.get("type").and_then(Value::as_str) != Some("report-artifact") {
        return DirectSubmitOutcome {
            success: true,
            summary: output.to_string(),
        };
    }

    let status = artifact
        .get("status")
        .and_then(Value::as_str)
        .unwrap_or("ok");
    let success = matches!(status, "ok" | "partial" | "empty");
    let report_name = artifact
        .get("report_name")
        .and_then(Value::as_str)
        .unwrap_or("report-artifact");
    let period = artifact
        .get("period")
        .and_then(Value::as_str)
        .unwrap_or("");
    let detail_rows = count_rows(artifact.get("counts"), artifact.get("rows"), "detail_rows");
    let summary_rows = count_summary_rows(artifact.get("counts"), artifact.get("sections"));
    let partial_reasons = artifact
        .get("partial_reasons")
        .and_then(Value::as_array)
        .map(|reasons| {
            reasons
                .iter()
                .filter_map(Value::as_str)
                .filter(|value| !value.trim().is_empty())
                .collect::<Vec<_>>()
        })
        .unwrap_or_default();

    let mut parts = vec![report_name.to_string()];
    if !period.trim().is_empty() {
        parts.push(period.to_string());
    }
    parts.push(format!("status={status}"));
    parts.push(format!("detail_rows={detail_rows}"));
    parts.push(format!("summary_rows={summary_rows}"));
    if !partial_reasons.is_empty() {
        parts.push(format!("partial_reasons={}", partial_reasons.join(",")));
    }

    DirectSubmitOutcome {
        success,
        summary: parts.join(" "),
    }
}

fn count_rows(counts: Option<&Value>, rows: Option<&Value>, key: &str) -> usize {
    counts
        .and_then(Value::as_object)
        .and_then(|counts| counts.get(key))
        .and_then(Value::as_u64)
        .map(|count| count as usize)
        .or_else(|| rows.and_then(Value::as_array).map(Vec::len))
        .unwrap_or(0)
}

fn count_summary_rows(counts: Option<&Value>, sections: Option<&Value>) -> usize {
    counts
        .and_then(Value::as_object)
        .and_then(|counts| counts.get("summary_rows"))
        .and_then(Value::as_u64)
        .map(|count| count as usize)
        .or_else(|| {
            sections
                .and_then(Value::as_array)
                .and_then(|sections| {
                    sections.iter().find_map(|section| {
                        section
                            .as_object()
                            .and_then(|section| section.get("rows"))
                            .and_then(Value::as_array)
                            .map(Vec::len)
                    })
                })
        })
        .unwrap_or(0)
}

fn parse_configured_tool_name(configured_tool: &str) -> Result<(&str, &str), PipeError> {
    let (skill_name, tool_name) = configured_tool.split_once('.').ok_or_else(|| {
        PipeError::Protocol(format!(
            "direct submit skill must use skill.tool format, got {configured_tool}"
        ))
    })?;
    let skill_name = skill_name.trim();
    let tool_name = tool_name.trim();
    if skill_name.is_empty() || tool_name.is_empty() {
        return Err(PipeError::Protocol(format!(
            "direct submit skill must use skill.tool format, got {configured_tool}"
        )));
    }
    Ok((skill_name, tool_name))
}

fn derive_expected_domain(task_context: &CompatTaskContext) -> Result<String, PipeError> {
    let page_url = task_context
        .page_url
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .ok_or_else(|| {
            PipeError::Protocol(
                "当前命令需要浏览器页面上下文才能执行。请在浏览器中打开目标页面后重试，或在指令末尾添加'。。。'使用确定性提交。".to_string(),
            )
        })?;

    Url::parse(page_url)
        .ok()
        .and_then(|url| url.host_str().map(|host| host.to_ascii_lowercase()))
        .ok_or_else(|| {
            PipeError::Protocol(format!(
                "direct submit skill could not derive expected_domain from page_url {page_url:?}"
            ))
        })
}

fn derive_period(instruction: &str) -> Result<String, PipeError> {
    let chars = instruction.chars().collect::<Vec<_>>();
    if chars.len() < 7 {
        return Err(PipeError::Protocol(
            "direct submit skill requires an explicit YYYY-MM period in the instruction"
                .to_string(),
        ));
    }

    for start in 0..=chars.len() - 7 {
        let candidate = chars[start..start + 7].iter().collect::<String>();
        if is_year_month(&candidate) {
            return Ok(candidate);
        }
    }

    Err(PipeError::Protocol(
        "direct submit skill requires an explicit YYYY-MM period in the instruction"
            .to_string(),
    ))
}

fn is_year_month(candidate: &str) -> bool {
    let bytes = candidate.as_bytes();
    bytes.len() == 7
        && bytes[0..4].iter().all(u8::is_ascii_digit)
        && bytes[4] == b'-'
        && bytes[5..7].iter().all(u8::is_ascii_digit)
        && matches!((bytes[5] - b'0') * 10 + (bytes[6] - b'0'), 1..=12)
}

#[cfg(test)]
mod tests {
    use super::{
        count_rows, count_summary_rows, derive_period, interpret_direct_submit_output,
        is_year_month, parse_configured_tool_name,
    };
    use serde_json::json;

    #[test]
    fn parse_configured_tool_name_requires_skill_and_tool() {
        assert_eq!(
            parse_configured_tool_name("fault-details-report.collect_fault_details")
                .unwrap(),
            ("fault-details-report", "collect_fault_details")
        );
        assert!(parse_configured_tool_name("fault-details-report").is_err());
    }

    #[test]
    fn derive_period_requires_explicit_year_month() {
        assert_eq!(derive_period("收集 2026-03 故障明细").unwrap(), "2026-03");
        assert!(derive_period("收集三月故障明细").is_err());
    }

    #[test]
    fn year_month_validation_rejects_invalid_month() {
        assert!(is_year_month("2026-12"));
        assert!(!is_year_month("2026-00"));
        assert!(!is_year_month("2026-13"));
    }

    #[test]
    fn interpret_direct_submit_output_maps_report_artifact_statuses() {
        let partial = interpret_direct_submit_output(
            &json!({
                "type": "report-artifact",
                "report_name": "fault-details-report",
                "period": "2026-03",
                "counts": { "detail_rows": 1, "summary_rows": 1 },
                "status": "partial",
                "partial_reasons": ["report_log_failed"]
            })
            .to_string(),
        );
        assert!(partial.success);
        assert!(partial.summary.contains("status=partial"));
        assert!(partial.summary.contains("report_log_failed"));

        let blocked = interpret_direct_submit_output(
            &json!({
                "type": "report-artifact",
                "report_name": "fault-details-report",
                "status": "blocked",
                "partial_reasons": ["selected_range_unavailable"]
            })
            .to_string(),
        );
        assert!(!blocked.success);
        assert!(blocked.summary.contains("status=blocked"));
    }

    #[test]
    fn row_count_helpers_fall_back_to_payload_shapes() {
        assert_eq!(
            count_rows(None, Some(&json!([{ "qxdbh": "QX-1" }, { "qxdbh": "QX-2" }])), "detail_rows"),
            2
        );
        assert_eq!(
            count_summary_rows(None, Some(&json!([{ "name": "summary-sheet", "rows": [{ "index": 1 }] }]))),
            1
        );
    }
}