// transcription.go -- the DashScope async task lifecycle (submit -> poll ->
// fetch result) and the channel-to-speaker mapping into the shared
// flyto.TranscriptionResponse. Wire shapes follow DashScope 录音文件识别:
// POST {base}/services/audio/asr/transcription with X-DashScope-Async:enable,
// GET {base}/tasks/{id} until SUCCEEDED/FAILED, then GET the per-file
// transcription_url (a temporary public JSON document).
//
// transcription.go -- DashScope 异步任务生命周期 (提交 -> 轮询 -> 取结果) 与
// 声道到说话人的映射. 线上形状按录音文件识别 API: 提交带 X-DashScope-Async,
// 轮询 tasks/{id} 到 SUCCEEDED/FAILED, 再 GET 每文件的 transcription_url
// (临时公开 JSON 文档).

package aliyunasr

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"sort"
	"strings"
	"time"

	"git.flytoex.net/yuanwei/flyto-agent/core/pkg/flyto"
)

// taskSubmitPath / taskQueryPath are appended to cfg.BaseURL.
// 提交/查询路径, 追加到 cfg.BaseURL.
const (
	taskSubmitPath = "/services/audio/asr/transcription"
	taskQueryPath  = "/tasks/"
)

// submitTask posts the async transcription task and returns its task_id.
// channel_id=[0,1] is sent when the caller labeled both stereo channels
// (LeftSpeaker+RightSpeaker -- the physical-channel diarization this backend
// exists for); otherwise the server default (merged) applies.
//
// submitTask 提交异步转写任务返回 task_id. 调用方标注了双声道
// (LeftSpeaker+RightSpeaker) 时带 channel_id=[0,1] (本后端存在的意义就是物理
// 声道分轨); 否则用服务端默认 (混轨).
func (p *Provider) submitTask(ctx context.Context, req *flyto.TranscriptionRequest, fileURL string) (string, error) {
	parameters := map[string]any{}
	if req.LeftSpeaker != "" && req.RightSpeaker != "" {
		parameters["channel_id"] = []int{0, 1}
	}
	body, err := json.Marshal(map[string]any{
		"model":      req.Model,
		"input":      map[string]any{"file_urls": []string{fileURL}},
		"parameters": parameters,
	})
	if err != nil {
		return "", fmt.Errorf("aliyunasr: marshal task: %w", err)
	}
	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, p.cfg.BaseURL+taskSubmitPath, bytes.NewReader(body))
	if err != nil {
		return "", fmt.Errorf("aliyunasr: build submit: %w", err)
	}
	httpReq.Header.Set("Authorization", "Bearer "+p.cfg.APIKey)
	httpReq.Header.Set("Content-Type", "application/json")
	// Async is opt-in per request; without this header the endpoint rejects
	// long jobs. 异步按请求开启; 不带此头长任务被拒.
	httpReq.Header.Set("X-DashScope-Async", "enable")

	var out struct {
		Output struct {
			TaskID     string `json:"task_id"`
			TaskStatus string `json:"task_status"`
		} `json:"output"`
		Code    string `json:"code"`
		Message string `json:"message"`
	}
	if err := p.doJSON(httpReq, &out); err != nil {
		return "", fmt.Errorf("aliyunasr: submit task: %w", err)
	}
	if out.Code != "" {
		return "", fmt.Errorf("aliyunasr: submit rejected: %s: %s", out.Code, out.Message)
	}
	if out.Output.TaskID == "" {
		return "", errors.New("aliyunasr: submit returned no task_id")
	}
	return out.Output.TaskID, nil
}

// pollTask polls the task until SUCCEEDED (returning the per-file result URL)
// or FAILED (fail loud with the server's reason). The loop is bounded by ctx:
// the caller's minutes-level deadline is the overall budget.
//
// pollTask 轮询任务到 SUCCEEDED (返回结果 URL) 或 FAILED (带服务端原因 fail
// loud). 循环受 ctx 界定: 调用方分钟级 deadline 即总预算.
func (p *Provider) pollTask(ctx context.Context, taskID string) (string, error) {
	for {
		httpReq, err := http.NewRequestWithContext(ctx, http.MethodGet, p.cfg.BaseURL+taskQueryPath+taskID, nil)
		if err != nil {
			return "", fmt.Errorf("aliyunasr: build poll: %w", err)
		}
		httpReq.Header.Set("Authorization", "Bearer "+p.cfg.APIKey)

		var out struct {
			Output struct {
				TaskStatus string `json:"task_status"`
				Results    []struct {
					TranscriptionURL string `json:"transcription_url"`
					SubTaskStatus    string `json:"subtask_status"`
					Code             string `json:"code"`
					Message          string `json:"message"`
				} `json:"results"`
				Code    string `json:"code"`
				Message string `json:"message"`
			} `json:"output"`
		}
		if err := p.doJSON(httpReq, &out); err != nil {
			return "", fmt.Errorf("aliyunasr: poll task %s: %w", taskID, err)
		}
		switch out.Output.TaskStatus {
		case "SUCCEEDED":
			if len(out.Output.Results) == 0 || out.Output.Results[0].TranscriptionURL == "" {
				return "", fmt.Errorf("aliyunasr: task %s succeeded but returned no transcription_url", taskID)
			}
			return out.Output.Results[0].TranscriptionURL, nil
		case "FAILED", "CANCELED", "UNKNOWN":
			reason := out.Output.Message
			if reason == "" && len(out.Output.Results) > 0 {
				reason = out.Output.Results[0].Message
			}
			return "", fmt.Errorf("aliyunasr: task %s %s: %s", taskID, strings.ToLower(out.Output.TaskStatus), reason)
		}
		select {
		case <-ctx.Done():
			return "", fmt.Errorf("aliyunasr: task %s not done before deadline: %w", taskID, ctx.Err())
		case <-time.After(p.pollEvery()):
		}
	}
}

// pollEvery returns the poll cadence (overridable in tests via pollOverride).
// 返回轮询间隔 (测试经 pollOverride 覆盖).
func (p *Provider) pollEvery() time.Duration {
	if p.cfg.pollOverride > 0 {
		return p.cfg.pollOverride
	}
	return pollInterval
}

// resultDocument is the JSON document behind transcription_url.
// resultDocument 是 transcription_url 背后的 JSON 文档.
type resultDocument struct {
	Properties struct {
		OriginalDurationMS int64 `json:"original_duration_in_milliseconds"`
	} `json:"properties"`
	Transcripts []struct {
		ChannelID int    `json:"channel_id"`
		Text      string `json:"text"`
		Sentences []struct {
			BeginTime int64  `json:"begin_time"`
			EndTime   int64  `json:"end_time"`
			Text      string `json:"text"`
		} `json:"sentences"`
	} `json:"transcripts"`
}

// fetchAndMapResult downloads the result document and maps it onto the shared
// contract: channel 0 -> LeftSpeaker, channel 1 -> RightSpeaker (fallback
// "channel_N" when unlabeled), per-channel sentences merged into one timeline
// sorted by begin_time, milliseconds -> seconds.
//
// fetchAndMapResult 下载结果文档并映射到共享契约: 声道 0 -> LeftSpeaker,
// 1 -> RightSpeaker (未标注回落 "channel_N"), 各声道 sentences 按 begin_time
// 合并成统一时间线, 毫秒转秒.
func (p *Provider) fetchAndMapResult(ctx context.Context, req *flyto.TranscriptionRequest, resultURL string) (*flyto.TranscriptionResponse, error) {
	httpReq, err := http.NewRequestWithContext(ctx, http.MethodGet, resultURL, nil)
	if err != nil {
		return nil, fmt.Errorf("aliyunasr: build result fetch: %w", err)
	}
	var doc resultDocument
	if err := p.doJSON(httpReq, &doc); err != nil {
		return nil, fmt.Errorf("aliyunasr: fetch result: %w", err)
	}

	speakerFor := func(channel int) string {
		switch channel {
		case 0:
			if req.LeftSpeaker != "" {
				return req.LeftSpeaker
			}
		case 1:
			if req.RightSpeaker != "" {
				return req.RightSpeaker
			}
		}
		if len(doc.Transcripts) <= 1 {
			return "" // single merged track: no speaker attribution. ZH: 单混轨不标说话人
		}
		return fmt.Sprintf("channel_%d", channel)
	}

	var segments []flyto.TranscriptionSegment
	for _, tr := range doc.Transcripts {
		speaker := speakerFor(tr.ChannelID)
		for _, s := range tr.Sentences {
			segments = append(segments, flyto.TranscriptionSegment{
				Start:   float64(s.BeginTime) / 1000,
				End:     float64(s.EndTime) / 1000,
				Text:    s.Text,
				Speaker: speaker,
			})
		}
	}
	sort.SliceStable(segments, func(i, j int) bool { return segments[i].Start < segments[j].Start })

	// Text: the interleaved timeline (speaker-prefixed when attributed) so a
	// text-only consumer still gets a readable conversation.
	// Text: 交叉时间线 (有归属时带说话人前缀), 纯文本消费者也能读出对话.
	var b strings.Builder
	for i, seg := range segments {
		if i > 0 {
			b.WriteByte('\n')
		}
		if seg.Speaker != "" {
			b.WriteString(seg.Speaker)
			b.WriteString(": ")
		}
		b.WriteString(seg.Text)
	}
	text := b.String()
	if text == "" {
		// No sentence detail: fall back to the per-channel full texts.
		// 无逐句明细: 回落各声道整段文本.
		var parts []string
		for _, tr := range doc.Transcripts {
			if tr.Text != "" {
				parts = append(parts, tr.Text)
			}
		}
		text = strings.Join(parts, "\n")
	}

	return &flyto.TranscriptionResponse{
		Text:     text,
		Duration: float64(doc.Properties.OriginalDurationMS) / 1000,
		Segments: segments,
	}, nil
}

// doJSON executes the request and decodes a JSON body, failing loud on
// non-2xx with a bounded body excerpt.
//
// doJSON 执行请求解码 JSON body, 非 2xx 带截断 body 摘要 fail loud.
func (p *Provider) doJSON(req *http.Request, out any) error {
	resp, err := p.client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
	if err != nil {
		return fmt.Errorf("read body: %w", err)
	}
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return fmt.Errorf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body[:min(len(body), 2048)])))
	}
	if err := json.Unmarshal(body, out); err != nil {
		return fmt.Errorf("decode JSON: %w", err)
	}
	return nil
}