channel.go 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977
  1. package model
  2. import (
  3. "database/sql/driver"
  4. "encoding/json"
  5. "errors"
  6. "fmt"
  7. "math/rand"
  8. "one-api/common"
  9. "one-api/constant"
  10. "one-api/dto"
  11. "one-api/types"
  12. "strings"
  13. "sync"
  14. "github.com/samber/lo"
  15. "gorm.io/gorm"
  16. )
  17. type Channel struct {
  18. Id int `json:"id"`
  19. Type int `json:"type" gorm:"default:0"`
  20. Key string `json:"key" gorm:"not null"`
  21. OpenAIOrganization *string `json:"openai_organization"`
  22. TestModel *string `json:"test_model"`
  23. Status int `json:"status" gorm:"default:1"`
  24. Name string `json:"name" gorm:"index"`
  25. Weight *uint `json:"weight" gorm:"default:0"`
  26. CreatedTime int64 `json:"created_time" gorm:"bigint"`
  27. TestTime int64 `json:"test_time" gorm:"bigint"`
  28. ResponseTime int `json:"response_time"` // in milliseconds
  29. BaseURL *string `json:"base_url" gorm:"column:base_url;default:''"`
  30. Other string `json:"other"`
  31. Balance float64 `json:"balance"` // in USD
  32. BalanceUpdatedTime int64 `json:"balance_updated_time" gorm:"bigint"`
  33. Models string `json:"models"`
  34. Group string `json:"group" gorm:"type:varchar(64);default:'default'"`
  35. UsedQuota int64 `json:"used_quota" gorm:"bigint;default:0"`
  36. ModelMapping *string `json:"model_mapping" gorm:"type:text"`
  37. //MaxInputTokens *int `json:"max_input_tokens" gorm:"default:0"`
  38. StatusCodeMapping *string `json:"status_code_mapping" gorm:"type:varchar(1024);default:''"`
  39. Priority *int64 `json:"priority" gorm:"bigint;default:0"`
  40. AutoBan *int `json:"auto_ban" gorm:"default:1"`
  41. OtherInfo string `json:"other_info"`
  42. OtherSettings string `json:"settings" gorm:"column:settings"` // 其他设置
  43. Tag *string `json:"tag" gorm:"index"`
  44. Setting *string `json:"setting" gorm:"type:text"` // 渠道额外设置
  45. ParamOverride *string `json:"param_override" gorm:"type:text"`
  46. HeaderOverride *string `json:"header_override" gorm:"type:text"`
  47. // add after v0.8.5
  48. ChannelInfo ChannelInfo `json:"channel_info" gorm:"type:json"`
  49. // cache info
  50. Keys []string `json:"-" gorm:"-"`
  51. }
  52. type ChannelInfo struct {
  53. IsMultiKey bool `json:"is_multi_key"` // 是否多Key模式
  54. MultiKeySize int `json:"multi_key_size"` // 多Key模式下的Key数量
  55. MultiKeyStatusList map[int]int `json:"multi_key_status_list"` // key状态列表,key index -> status
  56. MultiKeyDisabledReason map[int]string `json:"multi_key_disabled_reason,omitempty"` // key禁用原因列表,key index -> reason
  57. MultiKeyDisabledTime map[int]int64 `json:"multi_key_disabled_time,omitempty"` // key禁用时间列表,key index -> time
  58. MultiKeyPollingIndex int `json:"multi_key_polling_index"` // 多Key模式下轮询的key索引
  59. MultiKeyMode constant.MultiKeyMode `json:"multi_key_mode"`
  60. }
  61. // Value implements driver.Valuer interface
  62. func (c ChannelInfo) Value() (driver.Value, error) {
  63. return common.Marshal(&c)
  64. }
  65. // Scan implements sql.Scanner interface
  66. func (c *ChannelInfo) Scan(value interface{}) error {
  67. bytesValue, _ := value.([]byte)
  68. return common.Unmarshal(bytesValue, c)
  69. }
  70. func (channel *Channel) GetKeys() []string {
  71. if channel.Key == "" {
  72. return []string{}
  73. }
  74. if len(channel.Keys) > 0 {
  75. return channel.Keys
  76. }
  77. trimmed := strings.TrimSpace(channel.Key)
  78. // If the key starts with '[', try to parse it as a JSON array (e.g., for Vertex AI scenarios)
  79. if strings.HasPrefix(trimmed, "[") {
  80. var arr []json.RawMessage
  81. if err := common.Unmarshal([]byte(trimmed), &arr); err == nil {
  82. res := make([]string, len(arr))
  83. for i, v := range arr {
  84. res[i] = string(v)
  85. }
  86. return res
  87. }
  88. }
  89. // Otherwise, fall back to splitting by newline
  90. keys := strings.Split(strings.Trim(channel.Key, "\n"), "\n")
  91. return keys
  92. }
  93. func (channel *Channel) GetNextEnabledKey() (string, int, *types.NewAPIError) {
  94. // If not in multi-key mode, return the original key string directly.
  95. if !channel.ChannelInfo.IsMultiKey {
  96. return channel.Key, 0, nil
  97. }
  98. // Obtain all keys (split by \n)
  99. keys := channel.GetKeys()
  100. if len(keys) == 0 {
  101. // No keys available, return error, should disable the channel
  102. return "", 0, types.NewError(errors.New("no keys available"), types.ErrorCodeChannelNoAvailableKey)
  103. }
  104. statusList := channel.ChannelInfo.MultiKeyStatusList
  105. // helper to get key status, default to enabled when missing
  106. getStatus := func(idx int) int {
  107. if statusList == nil {
  108. return common.ChannelStatusEnabled
  109. }
  110. if status, ok := statusList[idx]; ok {
  111. return status
  112. }
  113. return common.ChannelStatusEnabled
  114. }
  115. // Collect indexes of enabled keys
  116. enabledIdx := make([]int, 0, len(keys))
  117. for i := range keys {
  118. if getStatus(i) == common.ChannelStatusEnabled {
  119. enabledIdx = append(enabledIdx, i)
  120. }
  121. }
  122. // If no specific status list or none enabled, fall back to first key
  123. if len(enabledIdx) == 0 {
  124. return keys[0], 0, nil
  125. }
  126. switch channel.ChannelInfo.MultiKeyMode {
  127. case constant.MultiKeyModeRandom:
  128. // Randomly pick one enabled key
  129. selectedIdx := enabledIdx[rand.Intn(len(enabledIdx))]
  130. return keys[selectedIdx], selectedIdx, nil
  131. case constant.MultiKeyModePolling:
  132. // Use channel-specific lock to ensure thread-safe polling
  133. lock := GetChannelPollingLock(channel.Id)
  134. lock.Lock()
  135. defer lock.Unlock()
  136. channelInfo, err := CacheGetChannelInfo(channel.Id)
  137. if err != nil {
  138. return "", 0, types.NewError(err, types.ErrorCodeGetChannelFailed, types.ErrOptionWithSkipRetry())
  139. }
  140. //println("before polling index:", channel.ChannelInfo.MultiKeyPollingIndex)
  141. defer func() {
  142. if common.DebugEnabled {
  143. println(fmt.Sprintf("channel %d polling index: %d", channel.Id, channel.ChannelInfo.MultiKeyPollingIndex))
  144. }
  145. if !common.MemoryCacheEnabled {
  146. _ = channel.SaveChannelInfo()
  147. } else {
  148. // CacheUpdateChannel(channel)
  149. }
  150. }()
  151. // Start from the saved polling index and look for the next enabled key
  152. start := channelInfo.MultiKeyPollingIndex
  153. if start < 0 || start >= len(keys) {
  154. start = 0
  155. }
  156. for i := 0; i < len(keys); i++ {
  157. idx := (start + i) % len(keys)
  158. if getStatus(idx) == common.ChannelStatusEnabled {
  159. // update polling index for next call (point to the next position)
  160. channel.ChannelInfo.MultiKeyPollingIndex = (idx + 1) % len(keys)
  161. return keys[idx], idx, nil
  162. }
  163. }
  164. // Fallback – should not happen, but return first enabled key
  165. return keys[enabledIdx[0]], enabledIdx[0], nil
  166. default:
  167. // Unknown mode, default to first enabled key (or original key string)
  168. return keys[enabledIdx[0]], enabledIdx[0], nil
  169. }
  170. }
  171. func (channel *Channel) SaveChannelInfo() error {
  172. return DB.Model(channel).Update("channel_info", channel.ChannelInfo).Error
  173. }
  174. func (channel *Channel) GetModels() []string {
  175. if channel.Models == "" {
  176. return []string{}
  177. }
  178. return strings.Split(strings.Trim(channel.Models, ","), ",")
  179. }
  180. func (channel *Channel) GetGroups() []string {
  181. if channel.Group == "" {
  182. return []string{}
  183. }
  184. groups := strings.Split(strings.Trim(channel.Group, ","), ",")
  185. for i, group := range groups {
  186. groups[i] = strings.TrimSpace(group)
  187. }
  188. return groups
  189. }
  190. func (channel *Channel) GetOtherInfo() map[string]interface{} {
  191. otherInfo := make(map[string]interface{})
  192. if channel.OtherInfo != "" {
  193. err := common.Unmarshal([]byte(channel.OtherInfo), &otherInfo)
  194. if err != nil {
  195. common.SysLog(fmt.Sprintf("failed to unmarshal other info: channel_id=%d, tag=%s, name=%s, error=%v", channel.Id, channel.GetTag(), channel.Name, err))
  196. }
  197. }
  198. return otherInfo
  199. }
  200. func (channel *Channel) SetOtherInfo(otherInfo map[string]interface{}) {
  201. otherInfoBytes, err := json.Marshal(otherInfo)
  202. if err != nil {
  203. common.SysLog(fmt.Sprintf("failed to marshal other info: channel_id=%d, tag=%s, name=%s, error=%v", channel.Id, channel.GetTag(), channel.Name, err))
  204. return
  205. }
  206. channel.OtherInfo = string(otherInfoBytes)
  207. }
  208. func (channel *Channel) GetTag() string {
  209. if channel.Tag == nil {
  210. return ""
  211. }
  212. return *channel.Tag
  213. }
  214. func (channel *Channel) SetTag(tag string) {
  215. channel.Tag = &tag
  216. }
  217. func (channel *Channel) GetAutoBan() bool {
  218. if channel.AutoBan == nil {
  219. return false
  220. }
  221. return *channel.AutoBan == 1
  222. }
  223. func (channel *Channel) Save() error {
  224. return DB.Save(channel).Error
  225. }
  226. func GetAllChannels(startIdx int, num int, selectAll bool, idSort bool) ([]*Channel, error) {
  227. var channels []*Channel
  228. var err error
  229. order := "priority desc"
  230. if idSort {
  231. order = "id desc"
  232. }
  233. if selectAll {
  234. err = DB.Order(order).Find(&channels).Error
  235. } else {
  236. err = DB.Order(order).Limit(num).Offset(startIdx).Omit("key").Find(&channels).Error
  237. }
  238. return channels, err
  239. }
  240. func GetChannelsByTag(tag string, idSort bool) ([]*Channel, error) {
  241. var channels []*Channel
  242. order := "priority desc"
  243. if idSort {
  244. order = "id desc"
  245. }
  246. err := DB.Where("tag = ?", tag).Order(order).Find(&channels).Error
  247. return channels, err
  248. }
  249. func SearchChannels(keyword string, group string, model string, idSort bool) ([]*Channel, error) {
  250. var channels []*Channel
  251. modelsCol := "`models`"
  252. // 如果是 PostgreSQL,使用双引号
  253. if common.UsingPostgreSQL {
  254. modelsCol = `"models"`
  255. }
  256. baseURLCol := "`base_url`"
  257. // 如果是 PostgreSQL,使用双引号
  258. if common.UsingPostgreSQL {
  259. baseURLCol = `"base_url"`
  260. }
  261. order := "priority desc"
  262. if idSort {
  263. order = "id desc"
  264. }
  265. // 构造基础查询
  266. baseQuery := DB.Model(&Channel{}).Omit("key")
  267. // 构造WHERE子句
  268. var whereClause string
  269. var args []interface{}
  270. if group != "" && group != "null" {
  271. var groupCondition string
  272. if common.UsingMySQL {
  273. groupCondition = `CONCAT(',', ` + commonGroupCol + `, ',') LIKE ?`
  274. } else {
  275. // sqlite, PostgreSQL
  276. groupCondition = `(',' || ` + commonGroupCol + ` || ',') LIKE ?`
  277. }
  278. whereClause = "(id = ? OR name LIKE ? OR " + commonKeyCol + " = ? OR " + baseURLCol + " LIKE ?) AND " + modelsCol + ` LIKE ? AND ` + groupCondition
  279. args = append(args, common.String2Int(keyword), "%"+keyword+"%", keyword, "%"+keyword+"%", "%"+model+"%", "%,"+group+",%")
  280. } else {
  281. whereClause = "(id = ? OR name LIKE ? OR " + commonKeyCol + " = ? OR " + baseURLCol + " LIKE ?) AND " + modelsCol + " LIKE ?"
  282. args = append(args, common.String2Int(keyword), "%"+keyword+"%", keyword, "%"+keyword+"%", "%"+model+"%")
  283. }
  284. // 执行查询
  285. err := baseQuery.Where(whereClause, args...).Order(order).Find(&channels).Error
  286. if err != nil {
  287. return nil, err
  288. }
  289. return channels, nil
  290. }
  291. func GetChannelById(id int, selectAll bool) (*Channel, error) {
  292. channel := &Channel{Id: id}
  293. var err error = nil
  294. if selectAll {
  295. err = DB.First(channel, "id = ?", id).Error
  296. } else {
  297. err = DB.Omit("key").First(channel, "id = ?", id).Error
  298. }
  299. if err != nil {
  300. return nil, err
  301. }
  302. if channel == nil {
  303. return nil, errors.New("channel not found")
  304. }
  305. return channel, nil
  306. }
  307. func BatchInsertChannels(channels []Channel) error {
  308. if len(channels) == 0 {
  309. return nil
  310. }
  311. tx := DB.Begin()
  312. if tx.Error != nil {
  313. return tx.Error
  314. }
  315. defer func() {
  316. if r := recover(); r != nil {
  317. tx.Rollback()
  318. }
  319. }()
  320. for _, chunk := range lo.Chunk(channels, 50) {
  321. if err := tx.Create(&chunk).Error; err != nil {
  322. tx.Rollback()
  323. return err
  324. }
  325. for _, channel_ := range chunk {
  326. if err := channel_.AddAbilities(tx); err != nil {
  327. tx.Rollback()
  328. return err
  329. }
  330. }
  331. }
  332. return tx.Commit().Error
  333. }
  334. func BatchDeleteChannels(ids []int) error {
  335. if len(ids) == 0 {
  336. return nil
  337. }
  338. // 使用事务 分批删除channel表和abilities表
  339. tx := DB.Begin()
  340. if tx.Error != nil {
  341. return tx.Error
  342. }
  343. for _, chunk := range lo.Chunk(ids, 200) {
  344. if err := tx.Where("id in (?)", chunk).Delete(&Channel{}).Error; err != nil {
  345. tx.Rollback()
  346. return err
  347. }
  348. if err := tx.Where("channel_id in (?)", chunk).Delete(&Ability{}).Error; err != nil {
  349. tx.Rollback()
  350. return err
  351. }
  352. }
  353. return tx.Commit().Error
  354. }
  355. func (channel *Channel) GetPriority() int64 {
  356. if channel.Priority == nil {
  357. return 0
  358. }
  359. return *channel.Priority
  360. }
  361. func (channel *Channel) GetWeight() int {
  362. if channel.Weight == nil {
  363. return 0
  364. }
  365. return int(*channel.Weight)
  366. }
  367. func (channel *Channel) GetBaseURL() string {
  368. if channel.BaseURL == nil {
  369. return ""
  370. }
  371. url := *channel.BaseURL
  372. if url == "" {
  373. url = constant.ChannelBaseURLs[channel.Type]
  374. }
  375. return url
  376. }
  377. func (channel *Channel) GetModelMapping() string {
  378. if channel.ModelMapping == nil {
  379. return ""
  380. }
  381. return *channel.ModelMapping
  382. }
  383. func (channel *Channel) GetStatusCodeMapping() string {
  384. if channel.StatusCodeMapping == nil {
  385. return ""
  386. }
  387. return *channel.StatusCodeMapping
  388. }
  389. func (channel *Channel) Insert() error {
  390. var err error
  391. err = DB.Create(channel).Error
  392. if err != nil {
  393. return err
  394. }
  395. err = channel.AddAbilities(nil)
  396. return err
  397. }
  398. func (channel *Channel) Update() error {
  399. // If this is a multi-key channel, recalculate MultiKeySize based on the current key list to avoid inconsistency after editing keys
  400. if channel.ChannelInfo.IsMultiKey {
  401. var keyStr string
  402. if channel.Key != "" {
  403. keyStr = channel.Key
  404. } else {
  405. // If key is not provided, read the existing key from the database
  406. if existing, err := GetChannelById(channel.Id, true); err == nil {
  407. keyStr = existing.Key
  408. }
  409. }
  410. // Parse the key list (supports newline separation or JSON array)
  411. keys := []string{}
  412. if keyStr != "" {
  413. trimmed := strings.TrimSpace(keyStr)
  414. if strings.HasPrefix(trimmed, "[") {
  415. var arr []json.RawMessage
  416. if err := common.Unmarshal([]byte(trimmed), &arr); err == nil {
  417. keys = make([]string, len(arr))
  418. for i, v := range arr {
  419. keys[i] = string(v)
  420. }
  421. }
  422. }
  423. if len(keys) == 0 { // fallback to newline split
  424. keys = strings.Split(strings.Trim(keyStr, "\n"), "\n")
  425. }
  426. }
  427. channel.ChannelInfo.MultiKeySize = len(keys)
  428. // Clean up status data that exceeds the new key count to prevent index out of range
  429. if channel.ChannelInfo.MultiKeyStatusList != nil {
  430. for idx := range channel.ChannelInfo.MultiKeyStatusList {
  431. if idx >= channel.ChannelInfo.MultiKeySize {
  432. delete(channel.ChannelInfo.MultiKeyStatusList, idx)
  433. }
  434. }
  435. }
  436. }
  437. var err error
  438. err = DB.Model(channel).Updates(channel).Error
  439. if err != nil {
  440. return err
  441. }
  442. DB.Model(channel).First(channel, "id = ?", channel.Id)
  443. err = channel.UpdateAbilities(nil)
  444. return err
  445. }
  446. func (channel *Channel) UpdateResponseTime(responseTime int64) {
  447. err := DB.Model(channel).Select("response_time", "test_time").Updates(Channel{
  448. TestTime: common.GetTimestamp(),
  449. ResponseTime: int(responseTime),
  450. }).Error
  451. if err != nil {
  452. common.SysLog(fmt.Sprintf("failed to update response time: channel_id=%d, error=%v", channel.Id, err))
  453. }
  454. }
  455. func (channel *Channel) UpdateBalance(balance float64) {
  456. err := DB.Model(channel).Select("balance_updated_time", "balance").Updates(Channel{
  457. BalanceUpdatedTime: common.GetTimestamp(),
  458. Balance: balance,
  459. }).Error
  460. if err != nil {
  461. common.SysLog(fmt.Sprintf("failed to update balance: channel_id=%d, error=%v", channel.Id, err))
  462. }
  463. }
  464. func (channel *Channel) Delete() error {
  465. var err error
  466. err = DB.Delete(channel).Error
  467. if err != nil {
  468. return err
  469. }
  470. err = channel.DeleteAbilities()
  471. return err
  472. }
  473. var channelStatusLock sync.Mutex
  474. // channelPollingLocks stores locks for each channel.id to ensure thread-safe polling
  475. var channelPollingLocks sync.Map
  476. // GetChannelPollingLock returns or creates a mutex for the given channel ID
  477. func GetChannelPollingLock(channelId int) *sync.Mutex {
  478. if lock, exists := channelPollingLocks.Load(channelId); exists {
  479. return lock.(*sync.Mutex)
  480. }
  481. // Create new lock for this channel
  482. newLock := &sync.Mutex{}
  483. actual, _ := channelPollingLocks.LoadOrStore(channelId, newLock)
  484. return actual.(*sync.Mutex)
  485. }
  486. // CleanupChannelPollingLocks removes locks for channels that no longer exist
  487. // This is optional and can be called periodically to prevent memory leaks
  488. func CleanupChannelPollingLocks() {
  489. var activeChannelIds []int
  490. DB.Model(&Channel{}).Pluck("id", &activeChannelIds)
  491. activeChannelSet := make(map[int]bool)
  492. for _, id := range activeChannelIds {
  493. activeChannelSet[id] = true
  494. }
  495. channelPollingLocks.Range(func(key, value interface{}) bool {
  496. channelId := key.(int)
  497. if !activeChannelSet[channelId] {
  498. channelPollingLocks.Delete(channelId)
  499. }
  500. return true
  501. })
  502. }
  503. func handlerMultiKeyUpdate(channel *Channel, usingKey string, status int, reason string) {
  504. keys := channel.GetKeys()
  505. if len(keys) == 0 {
  506. channel.Status = status
  507. } else {
  508. var keyIndex int
  509. for i, key := range keys {
  510. if key == usingKey {
  511. keyIndex = i
  512. break
  513. }
  514. }
  515. if channel.ChannelInfo.MultiKeyStatusList == nil {
  516. channel.ChannelInfo.MultiKeyStatusList = make(map[int]int)
  517. }
  518. if status == common.ChannelStatusEnabled {
  519. delete(channel.ChannelInfo.MultiKeyStatusList, keyIndex)
  520. } else {
  521. channel.ChannelInfo.MultiKeyStatusList[keyIndex] = status
  522. if channel.ChannelInfo.MultiKeyDisabledReason == nil {
  523. channel.ChannelInfo.MultiKeyDisabledReason = make(map[int]string)
  524. }
  525. if channel.ChannelInfo.MultiKeyDisabledTime == nil {
  526. channel.ChannelInfo.MultiKeyDisabledTime = make(map[int]int64)
  527. }
  528. channel.ChannelInfo.MultiKeyDisabledReason[keyIndex] = reason
  529. channel.ChannelInfo.MultiKeyDisabledTime[keyIndex] = common.GetTimestamp()
  530. }
  531. if len(channel.ChannelInfo.MultiKeyStatusList) >= channel.ChannelInfo.MultiKeySize {
  532. channel.Status = common.ChannelStatusAutoDisabled
  533. info := channel.GetOtherInfo()
  534. info["status_reason"] = "All keys are disabled"
  535. info["status_time"] = common.GetTimestamp()
  536. channel.SetOtherInfo(info)
  537. }
  538. }
  539. }
  540. func UpdateChannelStatus(channelId int, usingKey string, status int, reason string) bool {
  541. if common.MemoryCacheEnabled {
  542. channelStatusLock.Lock()
  543. defer channelStatusLock.Unlock()
  544. channelCache, _ := CacheGetChannel(channelId)
  545. if channelCache == nil {
  546. return false
  547. }
  548. if channelCache.ChannelInfo.IsMultiKey {
  549. // 如果是多Key模式,更新缓存中的状态
  550. handlerMultiKeyUpdate(channelCache, usingKey, status, reason)
  551. //CacheUpdateChannel(channelCache)
  552. //return true
  553. } else {
  554. // 如果缓存渠道存在,且状态已是目标状态,直接返回
  555. if channelCache.Status == status {
  556. return false
  557. }
  558. CacheUpdateChannelStatus(channelId, status)
  559. }
  560. }
  561. shouldUpdateAbilities := false
  562. defer func() {
  563. if shouldUpdateAbilities {
  564. err := UpdateAbilityStatus(channelId, status == common.ChannelStatusEnabled)
  565. if err != nil {
  566. common.SysLog(fmt.Sprintf("failed to update ability status: channel_id=%d, error=%v", channelId, err))
  567. }
  568. }
  569. }()
  570. channel, err := GetChannelById(channelId, true)
  571. if err != nil {
  572. return false
  573. } else {
  574. if channel.Status == status {
  575. return false
  576. }
  577. if channel.ChannelInfo.IsMultiKey {
  578. beforeStatus := channel.Status
  579. handlerMultiKeyUpdate(channel, usingKey, status, reason)
  580. if beforeStatus != channel.Status {
  581. shouldUpdateAbilities = true
  582. }
  583. } else {
  584. info := channel.GetOtherInfo()
  585. info["status_reason"] = reason
  586. info["status_time"] = common.GetTimestamp()
  587. channel.SetOtherInfo(info)
  588. channel.Status = status
  589. shouldUpdateAbilities = true
  590. }
  591. err = channel.Save()
  592. if err != nil {
  593. common.SysLog(fmt.Sprintf("failed to update channel status: channel_id=%d, status=%d, error=%v", channel.Id, status, err))
  594. return false
  595. }
  596. }
  597. return true
  598. }
  599. func EnableChannelByTag(tag string) error {
  600. err := DB.Model(&Channel{}).Where("tag = ?", tag).Update("status", common.ChannelStatusEnabled).Error
  601. if err != nil {
  602. return err
  603. }
  604. err = UpdateAbilityStatusByTag(tag, true)
  605. return err
  606. }
  607. func DisableChannelByTag(tag string) error {
  608. err := DB.Model(&Channel{}).Where("tag = ?", tag).Update("status", common.ChannelStatusManuallyDisabled).Error
  609. if err != nil {
  610. return err
  611. }
  612. err = UpdateAbilityStatusByTag(tag, false)
  613. return err
  614. }
  615. func EditChannelByTag(tag string, newTag *string, modelMapping *string, models *string, group *string, priority *int64, weight *uint) error {
  616. updateData := Channel{}
  617. shouldReCreateAbilities := false
  618. updatedTag := tag
  619. // 如果 newTag 不为空且不等于 tag,则更新 tag
  620. if newTag != nil && *newTag != tag {
  621. updateData.Tag = newTag
  622. updatedTag = *newTag
  623. }
  624. if modelMapping != nil && *modelMapping != "" {
  625. updateData.ModelMapping = modelMapping
  626. }
  627. if models != nil && *models != "" {
  628. shouldReCreateAbilities = true
  629. updateData.Models = *models
  630. }
  631. if group != nil && *group != "" {
  632. shouldReCreateAbilities = true
  633. updateData.Group = *group
  634. }
  635. if priority != nil {
  636. updateData.Priority = priority
  637. }
  638. if weight != nil {
  639. updateData.Weight = weight
  640. }
  641. err := DB.Model(&Channel{}).Where("tag = ?", tag).Updates(updateData).Error
  642. if err != nil {
  643. return err
  644. }
  645. if shouldReCreateAbilities {
  646. channels, err := GetChannelsByTag(updatedTag, false)
  647. if err == nil {
  648. for _, channel := range channels {
  649. err = channel.UpdateAbilities(nil)
  650. if err != nil {
  651. common.SysLog(fmt.Sprintf("failed to update abilities: channel_id=%d, tag=%s, error=%v", channel.Id, channel.GetTag(), err))
  652. }
  653. }
  654. }
  655. } else {
  656. err := UpdateAbilityByTag(tag, newTag, priority, weight)
  657. if err != nil {
  658. return err
  659. }
  660. }
  661. return nil
  662. }
  663. func UpdateChannelUsedQuota(id int, quota int) {
  664. if common.BatchUpdateEnabled {
  665. addNewRecord(BatchUpdateTypeChannelUsedQuota, id, quota)
  666. return
  667. }
  668. updateChannelUsedQuota(id, quota)
  669. }
  670. func updateChannelUsedQuota(id int, quota int) {
  671. err := DB.Model(&Channel{}).Where("id = ?", id).Update("used_quota", gorm.Expr("used_quota + ?", quota)).Error
  672. if err != nil {
  673. common.SysLog(fmt.Sprintf("failed to update channel used quota: channel_id=%d, delta_quota=%d, error=%v", id, quota, err))
  674. }
  675. }
  676. func DeleteChannelByStatus(status int64) (int64, error) {
  677. result := DB.Where("status = ?", status).Delete(&Channel{})
  678. return result.RowsAffected, result.Error
  679. }
  680. func DeleteDisabledChannel() (int64, error) {
  681. result := DB.Where("status = ? or status = ?", common.ChannelStatusAutoDisabled, common.ChannelStatusManuallyDisabled).Delete(&Channel{})
  682. return result.RowsAffected, result.Error
  683. }
  684. func GetPaginatedTags(offset int, limit int) ([]*string, error) {
  685. var tags []*string
  686. err := DB.Model(&Channel{}).Select("DISTINCT tag").Where("tag != ''").Offset(offset).Limit(limit).Find(&tags).Error
  687. return tags, err
  688. }
  689. func SearchTags(keyword string, group string, model string, idSort bool) ([]*string, error) {
  690. var tags []*string
  691. modelsCol := "`models`"
  692. // 如果是 PostgreSQL,使用双引号
  693. if common.UsingPostgreSQL {
  694. modelsCol = `"models"`
  695. }
  696. baseURLCol := "`base_url`"
  697. // 如果是 PostgreSQL,使用双引号
  698. if common.UsingPostgreSQL {
  699. baseURLCol = `"base_url"`
  700. }
  701. order := "priority desc"
  702. if idSort {
  703. order = "id desc"
  704. }
  705. // 构造基础查询
  706. baseQuery := DB.Model(&Channel{}).Omit("key")
  707. // 构造WHERE子句
  708. var whereClause string
  709. var args []interface{}
  710. if group != "" && group != "null" {
  711. var groupCondition string
  712. if common.UsingMySQL {
  713. groupCondition = `CONCAT(',', ` + commonGroupCol + `, ',') LIKE ?`
  714. } else {
  715. // sqlite, PostgreSQL
  716. groupCondition = `(',' || ` + commonGroupCol + ` || ',') LIKE ?`
  717. }
  718. whereClause = "(id = ? OR name LIKE ? OR " + commonKeyCol + " = ? OR " + baseURLCol + " LIKE ?) AND " + modelsCol + ` LIKE ? AND ` + groupCondition
  719. args = append(args, common.String2Int(keyword), "%"+keyword+"%", keyword, "%"+keyword+"%", "%"+model+"%", "%,"+group+",%")
  720. } else {
  721. whereClause = "(id = ? OR name LIKE ? OR " + commonKeyCol + " = ? OR " + baseURLCol + " LIKE ?) AND " + modelsCol + " LIKE ?"
  722. args = append(args, common.String2Int(keyword), "%"+keyword+"%", keyword, "%"+keyword+"%", "%"+model+"%")
  723. }
  724. subQuery := baseQuery.Where(whereClause, args...).
  725. Select("tag").
  726. Where("tag != ''").
  727. Order(order)
  728. err := DB.Table("(?) as sub", subQuery).
  729. Select("DISTINCT tag").
  730. Find(&tags).Error
  731. if err != nil {
  732. return nil, err
  733. }
  734. return tags, nil
  735. }
  736. func (channel *Channel) ValidateSettings() error {
  737. channelParams := &dto.ChannelSettings{}
  738. if channel.Setting != nil && *channel.Setting != "" {
  739. err := common.Unmarshal([]byte(*channel.Setting), channelParams)
  740. if err != nil {
  741. return err
  742. }
  743. }
  744. return nil
  745. }
  746. func (channel *Channel) GetSetting() dto.ChannelSettings {
  747. setting := dto.ChannelSettings{}
  748. if channel.Setting != nil && *channel.Setting != "" {
  749. err := common.Unmarshal([]byte(*channel.Setting), &setting)
  750. if err != nil {
  751. common.SysLog(fmt.Sprintf("failed to unmarshal setting: channel_id=%d, error=%v", channel.Id, err))
  752. channel.Setting = nil // 清空设置以避免后续错误
  753. _ = channel.Save() // 保存修改
  754. }
  755. }
  756. return setting
  757. }
  758. func (channel *Channel) SetSetting(setting dto.ChannelSettings) {
  759. settingBytes, err := common.Marshal(setting)
  760. if err != nil {
  761. common.SysLog(fmt.Sprintf("failed to marshal setting: channel_id=%d, error=%v", channel.Id, err))
  762. return
  763. }
  764. channel.Setting = common.GetPointer[string](string(settingBytes))
  765. }
  766. func (channel *Channel) GetOtherSettings() dto.ChannelOtherSettings {
  767. setting := dto.ChannelOtherSettings{}
  768. if channel.OtherSettings != "" {
  769. err := common.UnmarshalJsonStr(channel.OtherSettings, &setting)
  770. if err != nil {
  771. common.SysLog(fmt.Sprintf("failed to unmarshal setting: channel_id=%d, error=%v", channel.Id, err))
  772. channel.OtherSettings = "{}" // 清空设置以避免后续错误
  773. _ = channel.Save() // 保存修改
  774. }
  775. }
  776. return setting
  777. }
  778. func (channel *Channel) SetOtherSettings(setting dto.ChannelOtherSettings) {
  779. settingBytes, err := common.Marshal(setting)
  780. if err != nil {
  781. common.SysLog(fmt.Sprintf("failed to marshal setting: channel_id=%d, error=%v", channel.Id, err))
  782. return
  783. }
  784. channel.OtherSettings = string(settingBytes)
  785. }
  786. func (channel *Channel) GetParamOverride() map[string]interface{} {
  787. paramOverride := make(map[string]interface{})
  788. if channel.ParamOverride != nil && *channel.ParamOverride != "" {
  789. err := common.Unmarshal([]byte(*channel.ParamOverride), &paramOverride)
  790. if err != nil {
  791. common.SysLog(fmt.Sprintf("failed to unmarshal param override: channel_id=%d, error=%v", channel.Id, err))
  792. }
  793. }
  794. return paramOverride
  795. }
  796. func (channel *Channel) GetHeaderOverride() map[string]interface{} {
  797. headerOverride := make(map[string]interface{})
  798. if channel.HeaderOverride != nil && *channel.HeaderOverride != "" {
  799. err := common.Unmarshal([]byte(*channel.HeaderOverride), &headerOverride)
  800. if err != nil {
  801. common.SysLog(fmt.Sprintf("failed to unmarshal param override: channel_id=%d, error=%v", channel.Id, err))
  802. }
  803. }
  804. return headerOverride
  805. }
  806. func GetChannelsByIds(ids []int) ([]*Channel, error) {
  807. var channels []*Channel
  808. err := DB.Where("id in (?)", ids).Find(&channels).Error
  809. return channels, err
  810. }
  811. func BatchSetChannelTag(ids []int, tag *string) error {
  812. // 开启事务
  813. tx := DB.Begin()
  814. if tx.Error != nil {
  815. return tx.Error
  816. }
  817. // 更新标签
  818. err := tx.Model(&Channel{}).Where("id in (?)", ids).Update("tag", tag).Error
  819. if err != nil {
  820. tx.Rollback()
  821. return err
  822. }
  823. // update ability status
  824. channels, err := GetChannelsByIds(ids)
  825. if err != nil {
  826. tx.Rollback()
  827. return err
  828. }
  829. for _, channel := range channels {
  830. err = channel.UpdateAbilities(tx)
  831. if err != nil {
  832. tx.Rollback()
  833. return err
  834. }
  835. }
  836. // 提交事务
  837. return tx.Commit().Error
  838. }
  839. // CountAllChannels returns total channels in DB
  840. func CountAllChannels() (int64, error) {
  841. var total int64
  842. err := DB.Model(&Channel{}).Count(&total).Error
  843. return total, err
  844. }
  845. // CountAllTags returns number of non-empty distinct tags
  846. func CountAllTags() (int64, error) {
  847. var total int64
  848. err := DB.Model(&Channel{}).Where("tag is not null AND tag != ''").Distinct("tag").Count(&total).Error
  849. return total, err
  850. }
  851. // Get channels of specified type with pagination
  852. func GetChannelsByType(startIdx int, num int, idSort bool, channelType int) ([]*Channel, error) {
  853. var channels []*Channel
  854. order := "priority desc"
  855. if idSort {
  856. order = "id desc"
  857. }
  858. err := DB.Where("type = ?", channelType).Order(order).Limit(num).Offset(startIdx).Omit("key").Find(&channels).Error
  859. return channels, err
  860. }
  861. // Count channels of specific type
  862. func CountChannelsByType(channelType int) (int64, error) {
  863. var count int64
  864. err := DB.Model(&Channel{}).Where("type = ?", channelType).Count(&count).Error
  865. return count, err
  866. }
  867. // Return map[type]count for all channels
  868. func CountChannelsGroupByType() (map[int64]int64, error) {
  869. type result struct {
  870. Type int64 `gorm:"column:type"`
  871. Count int64 `gorm:"column:count"`
  872. }
  873. var results []result
  874. err := DB.Model(&Channel{}).Select("type, count(*) as count").Group("type").Find(&results).Error
  875. if err != nil {
  876. return nil, err
  877. }
  878. counts := make(map[int64]int64)
  879. for _, r := range results {
  880. counts[r.Type] = r.Count
  881. }
  882. return counts, nil
  883. }