// Package aliyunasr implements flyto.TranscriptionProvider against Aliyun
// DashScope's async file-transcription API (录音文件识别, model
// paraformer-8k-v2) -- the cloud fallback for the self-hosted oMLX ASR box.
//
// The wire protocol differs from the OpenAI-compatible path in three ways the
// provider absorbs so consumers see ZERO difference (same TranscriptionRequest
// in, same TranscriptionResponse out):
//
//  1. ASYNC: DashScope takes a task submission and is polled until SUCCEEDED;
//     Transcribe polls internally and returns once, preserving the synchronous
//     contract (callers already budget minutes-level deadlines).
//  2. URL INPUT: the task API only accepts a publicly reachable file URL, not
//     bytes. Transcribe spools the audio stream to a temp file (size needed
//     for the upload; disk, never memory), PUTs it to an OSS bucket with a V1
//     header signature, hands the task a time-limited signed GET URL, and
//     best-effort deletes the object afterwards (a bucket lifecycle rule is
//     the real cleaner).
//  3. CHANNEL SPLIT: channel_id=[0,1] makes DashScope transcribe each stereo
//     channel independently -- the physical channel IS the speaker, so
//     channel 0 maps to req.LeftSpeaker and channel 1 to req.RightSpeaker,
//     and the per-channel sentences are merged into one segment timeline
//     (begin_time ascending, milliseconds converted to the contract's
//     seconds).
//
// Chat is NOT supported: Stream fails loud (transcription-only provider),
// mirroring how consumers type-assert optional capabilities.
//
// aliyunasr 包对阿里云 DashScope 异步录音文件识别 API (paraformer-8k-v2) 实现
// flyto.TranscriptionProvider -- 自托管 oMLX ASR 的云端备选. 线协议与 OpenAI
// 兼容路径的三个差异全部在 provider 内吸收, 消费者零感知 (同请求进, 同响应出):
// 1. 异步: 提交任务 + 内部轮询到 SUCCEEDED 一次性返回, 保持同步语义 (调用方
// 本来就按分钟级预算 deadline). 2. URL 输入: 任务只收公网 URL, 先落临时盘
// (要长度; 走盘不走内存), V1 头签名 PUT 进 OSS, 给任务限时签名 GET URL, 完成后
// 尽力删除 (真正的清理靠 bucket 生命周期规则). 3. 声道分轨: channel_id=[0,1]
// 让左右声道各出一路独立转写 -- 物理声道即说话人, 0 -> LeftSpeaker, 1 ->
// RightSpeaker, 两路 sentences 按 begin_time 合并成统一时间线 (毫秒转秒).
// 不支持 chat: Stream fail-loud (纯转写 provider).
package aliyunasr

import (
	"context"
	"crypto/hmac"
	"crypto/rand"
	"crypto/sha1"
	"encoding/base64"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"net/http"
	"os"
	"strings"
	"time"

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

// DefaultBaseURL is the public DashScope API root; a workspace-scoped
// deployment (maas.aliyuncs.com host) overrides it via Config.BaseURL.
//
// DefaultBaseURL 是公共 DashScope API 根; 业务空间专属部署 (maas.aliyuncs.com
// host) 经 Config.BaseURL 覆盖.
const DefaultBaseURL = "https://dashscope.aliyuncs.com/api/v1"

// ModelParaformer8kV2 is the 8k telephony-tuned file-transcription model this
// provider is validated against (the only entry in Models()).
//
// ModelParaformer8kV2 是本 provider 实测过的 8k 电话音质录音识别模型 (Models()
// 唯一条目).
const ModelParaformer8kV2 = "paraformer-8k-v2"

// pollInterval paces the task-status polling loop; the overall budget is the
// caller's ctx deadline. 轮询间隔; 总预算是调用方 ctx deadline.
const pollInterval = 4 * time.Second

// OSSConfig locates the staging bucket used to hand DashScope a URL. Endpoint
// is the BUCKET base URL (virtual-host style,
// e.g. https://flytocall-audio.oss-cn-shanghai.aliyuncs.com); Bucket is the
// bucket name used in the V1 signature resource string. Empty credentials are
// allowed at construction and fail loud at Transcribe -- an instance can be
// registered before the operator provisions the sub-account key.
//
// OSSConfig 定位中转 bucket. Endpoint 是 BUCKET 级 base URL (虚拟主机式);
// Bucket 是 V1 签名资源串里的桶名. 凭据允许构造期为空, Transcribe 时 fail
// loud -- 实例可以先注册, 子账号 key 后配.
type OSSConfig struct {
	Endpoint        string
	Bucket          string
	AccessKeyID     string
	AccessKeySecret string
}

// Config configures the provider. APIKey is the DashScope key (required at
// use); BaseURL empty -> DefaultBaseURL; HTTPClient nil -> a 5-minute-timeout
// client (individual polls are short; the long wait lives in the poll LOOP
// bounded by ctx, not in any single request).
//
// Config 配置 provider. APIKey 是 DashScope key (使用时必填); BaseURL 空 ->
// DefaultBaseURL; HTTPClient nil -> 5 分钟超时 client (单次轮询很短; 长等待在
// ctx 界定的轮询循环里, 不在单个请求上).
type Config struct {
	APIKey     string
	BaseURL    string
	OSS        OSSConfig
	HTTPClient *http.Client

	// pollOverride shortens the task poll cadence in tests (0 = production
	// pollInterval). Unexported: not a knob operators should reach for.
	// pollOverride 供测试缩短轮询间隔 (0 = 生产 pollInterval). 不导出: 不是
	// 运维该碰的旋钮.
	pollOverride time.Duration
}

// Provider implements flyto.ModelProvider (Stream fails loud) and
// flyto.TranscriptionProvider (the real capability).
//
// Provider 实现 flyto.ModelProvider (Stream fail-loud) 与
// flyto.TranscriptionProvider (真能力).
type Provider struct {
	cfg    Config
	client *http.Client
}

var (
	_ flyto.ModelProvider         = (*Provider)(nil)
	_ flyto.TranscriptionProvider = (*Provider)(nil)
)

// New constructs the provider. 构造 provider.
func New(cfg Config) *Provider {
	if cfg.BaseURL == "" {
		cfg.BaseURL = DefaultBaseURL
	}
	cfg.BaseURL = strings.TrimRight(cfg.BaseURL, "/")
	cfg.OSS.Endpoint = strings.TrimRight(cfg.OSS.Endpoint, "/")
	client := cfg.HTTPClient
	if client == nil {
		client = &http.Client{Timeout: 5 * time.Minute}
	}
	return &Provider{cfg: cfg, client: client}
}

// Name implements flyto.ModelProvider. 实现 flyto.ModelProvider.
func (p *Provider) Name() string { return "aliyun-asr" }

// Stream implements flyto.ModelProvider by failing loud: this is a
// transcription-only provider, chat routing to it is a configuration error.
//
// Stream 以 fail-loud 实现 flyto.ModelProvider: 纯转写 provider, chat 路由到
// 这里是配置错误.
func (p *Provider) Stream(ctx context.Context, req *flyto.Request) (<-chan flyto.Event, error) {
	return nil, errors.New("aliyunasr: transcription-only provider; chat is not supported (route chat roles to a chat-capable instance)")
}

// Models implements flyto.ModelProvider with the static validated catalog.
// Models 返回静态实测目录.
func (p *Provider) Models(ctx context.Context) ([]flyto.ModelInfo, error) {
	return []flyto.ModelInfo{{
		ID:                    ModelParaformer8kV2,
		DisplayName:           "Paraformer 8k v2 (阿里云录音文件识别)",
		Provider:              p.Name(),
		SupportsTranscription: true,
	}}, nil
}

// Transcribe implements flyto.TranscriptionProvider: spool -> OSS PUT ->
// submit task -> poll -> fetch result -> map channels to speakers. Blocks
// until done; budget the ctx in minutes for long recordings.
//
// Transcribe 实现 flyto.TranscriptionProvider: 落盘 -> OSS PUT -> 提任务 ->
// 轮询 -> 取结果 -> 声道映射说话人. 阻塞到完成; 长录音按分钟级预算 ctx.
func (p *Provider) Transcribe(ctx context.Context, req *flyto.TranscriptionRequest) (*flyto.TranscriptionResponse, error) {
	if req == nil || req.Audio == nil {
		return nil, errors.New("aliyunasr: audio stream is required")
	}
	if req.Model == "" {
		return nil, errors.New("aliyunasr: model is required")
	}
	if p.cfg.APIKey == "" {
		return nil, errors.New("aliyunasr: DashScope api key is not configured")
	}
	if p.cfg.OSS.AccessKeyID == "" || p.cfg.OSS.AccessKeySecret == "" || p.cfg.OSS.Endpoint == "" || p.cfg.OSS.Bucket == "" {
		return nil, errors.New("aliyunasr: OSS staging is not configured (endpoint/bucket/access key required; set ALIBABA_CLOUD_ACCESS_KEY_ID / ALIBABA_CLOUD_ACCESS_KEY_SECRET)")
	}

	objectKey, size, tmpPath, err := spoolToTemp(req.Audio, req.Filename)
	if err != nil {
		return nil, err
	}
	defer os.Remove(tmpPath)

	if err := p.ossPut(ctx, objectKey, tmpPath, size); err != nil {
		return nil, err
	}
	// Best-effort delete: the lifecycle rule is the real cleaner, this just
	// keeps the bucket tidy on the happy path.
	// 尽力删除: 真清理靠生命周期规则, 这里只是让 happy path 干净.
	defer p.ossDelete(objectKey)

	fileURL := p.ossSignedGetURL(objectKey, time.Now().Add(2*time.Hour))

	taskID, err := p.submitTask(ctx, req, fileURL)
	if err != nil {
		return nil, err
	}
	resultURL, err := p.pollTask(ctx, taskID)
	if err != nil {
		return nil, err
	}
	return p.fetchAndMapResult(ctx, req, resultURL)
}

// spoolToTemp drains the audio stream to a temp file (DISK, never memory: the
// OSS PUT needs a Content-Length and recordings run to hundreds of MB) and
// mints a collision-safe object key carrying the filename hint.
//
// spoolToTemp 把音频流落到临时文件 (走盘不走内存: OSS PUT 要 Content-Length,
// 录音可达数百 MB), 并生成带文件名提示的防撞对象 key.
func spoolToTemp(audio io.Reader, filename string) (objectKey string, size int64, tmpPath string, err error) {
	f, err := os.CreateTemp("", "aliyunasr-*")
	if err != nil {
		return "", 0, "", fmt.Errorf("aliyunasr: create spool file: %w", err)
	}
	size, err = io.Copy(f, audio)
	closeErr := f.Close()
	if err != nil {
		os.Remove(f.Name())
		return "", 0, "", fmt.Errorf("aliyunasr: spool audio: %w", err)
	}
	if closeErr != nil {
		os.Remove(f.Name())
		return "", 0, "", fmt.Errorf("aliyunasr: close spool file: %w", closeErr)
	}
	if filename == "" {
		filename = "audio"
	}
	var rnd [8]byte
	if _, err := rand.Read(rnd[:]); err != nil {
		os.Remove(f.Name())
		return "", 0, "", fmt.Errorf("aliyunasr: object key entropy: %w", err)
	}
	// Path-safe key: strip any client-supplied directory components.
	// 路径安全 key: 去掉客户端带的目录成分.
	base := filename[strings.LastIndexByte(filename, '/')+1:]
	objectKey = fmt.Sprintf("transcribe/%d-%s-%s", time.Now().UnixMilli(), hex.EncodeToString(rnd[:]), base)
	return objectKey, size, f.Name(), nil
}

// ossSign computes the OSS V1 signature over the canonical string.
// ossSign 计算 OSS V1 签名.
func (p *Provider) ossSign(stringToSign string) string {
	mac := hmac.New(sha1.New, []byte(p.cfg.OSS.AccessKeySecret))
	mac.Write([]byte(stringToSign))
	return base64.StdEncoding.EncodeToString(mac.Sum(nil))
}

// ossPut uploads the spooled file with a V1 header-signed PUT.
// ossPut 用 V1 头签名 PUT 上传落盘文件.
func (p *Provider) ossPut(ctx context.Context, objectKey, tmpPath string, size int64) error {
	f, err := os.Open(tmpPath)
	if err != nil {
		return fmt.Errorf("aliyunasr: reopen spool file: %w", err)
	}
	defer f.Close()

	const contentType = "application/octet-stream"
	date := time.Now().UTC().Format(http.TimeFormat)
	resource := "/" + p.cfg.OSS.Bucket + "/" + objectKey
	sig := p.ossSign("PUT\n\n" + contentType + "\n" + date + "\n" + resource)

	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPut, p.cfg.OSS.Endpoint+"/"+objectKey, f)
	if err != nil {
		return fmt.Errorf("aliyunasr: build OSS PUT: %w", err)
	}
	httpReq.ContentLength = size
	httpReq.Header.Set("Content-Type", contentType)
	httpReq.Header.Set("Date", date)
	httpReq.Header.Set("Authorization", "OSS "+p.cfg.OSS.AccessKeyID+":"+sig)

	resp, err := p.client.Do(httpReq)
	if err != nil {
		return fmt.Errorf("aliyunasr: OSS upload: %w", err)
	}
	defer resp.Body.Close()
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		body, _ := io.ReadAll(io.LimitReader(resp.Body, 2048))
		return fmt.Errorf("aliyunasr: OSS upload failed: HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
	}
	return nil
}

// ossDelete removes the staged object, best-effort (background context: runs
// in a defer after the caller's ctx may already be done).
//
// ossDelete 尽力删除中转对象 (background context: 在 defer 里跑, 调用方 ctx
// 可能已结束).
func (p *Provider) ossDelete(objectKey string) {
	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()
	date := time.Now().UTC().Format(http.TimeFormat)
	resource := "/" + p.cfg.OSS.Bucket + "/" + objectKey
	sig := p.ossSign("DELETE\n\n\n" + date + "\n" + resource)
	httpReq, err := http.NewRequestWithContext(ctx, http.MethodDelete, p.cfg.OSS.Endpoint+"/"+objectKey, nil)
	if err != nil {
		return
	}
	httpReq.Header.Set("Date", date)
	httpReq.Header.Set("Authorization", "OSS "+p.cfg.OSS.AccessKeyID+":"+sig)
	if resp, err := p.client.Do(httpReq); err == nil {
		resp.Body.Close()
	}
}

// ossSignedGetURL builds a V1 query-signed, time-limited GET URL DashScope
// can fetch without credentials.
//
// ossSignedGetURL 构造 V1 查询签名限时 GET URL, DashScope 无凭据可取.
func (p *Provider) ossSignedGetURL(objectKey string, expiresAt time.Time) string {
	expires := fmt.Sprintf("%d", expiresAt.Unix())
	resource := "/" + p.cfg.OSS.Bucket + "/" + objectKey
	sig := p.ossSign("GET\n\n\n" + expires + "\n" + resource)
	q := "OSSAccessKeyId=" + urlQueryEscape(p.cfg.OSS.AccessKeyID) +
		"&Expires=" + expires +
		"&Signature=" + urlQueryEscape(sig)
	return p.cfg.OSS.Endpoint + "/" + objectKey + "?" + q
}

// urlQueryEscape percent-encodes a query value (stdlib net/url QueryEscape
// encodes space as '+', which OSS rejects in signatures; use %20 semantics).
//
// urlQueryEscape 百分号编码查询值 (net/url QueryEscape 把空格编成 '+', OSS
// 签名不认; 用 %20 语义).
func urlQueryEscape(s string) string {
	const hexDigits = "0123456789ABCDEF"
	var b strings.Builder
	for i := 0; i < len(s); i++ {
		c := s[i]
		if (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') ||
			c == '-' || c == '_' || c == '.' || c == '~' {
			b.WriteByte(c)
			continue
		}
		b.WriteByte('%')
		b.WriteByte(hexDigits[c>>4])
		b.WriteByte(hexDigits[c&0xF])
	}
	return b.String()
}