client.go 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821
  1. package telegram
  2. import (
  3. "bufio"
  4. "context"
  5. "encoding/base64"
  6. "fmt"
  7. "log"
  8. "math/rand/v2"
  9. "net"
  10. "net/url"
  11. "regexp"
  12. "sort"
  13. "strings"
  14. "sync"
  15. "time"
  16. "github.com/gotd/td/session"
  17. "github.com/gotd/td/telegram"
  18. "github.com/gotd/td/telegram/dcs"
  19. "github.com/gotd/td/tg"
  20. "github.com/gotd/td/tgerr"
  21. xproxy "golang.org/x/net/proxy"
  22. )
  23. var tmeRegexp = regexp.MustCompile(`https?://t\.me/[^\s"'<>)\]]+`)
  24. // Client TG 客户端
  25. type Client struct {
  26. account Account
  27. sessionPath string
  28. proxyURL string // SOCKS5/HTTP proxy URL
  29. mu sync.Mutex
  30. tgc *telegram.Client
  31. api *tg.Client
  32. cancel context.CancelFunc
  33. ready chan struct{} // closed when connected
  34. runErr error
  35. }
  36. // New 创建客户端(不连接,只初始化)
  37. func New(account Account) *Client {
  38. return &Client{
  39. account: account,
  40. sessionPath: account.SessionFile,
  41. ready: make(chan struct{}),
  42. }
  43. }
  44. // SetProxy sets the proxy URL for this client's connections.
  45. func (c *Client) SetProxy(proxyURL string) {
  46. c.mu.Lock()
  47. c.proxyURL = proxyURL
  48. c.mu.Unlock()
  49. }
  50. // Connect 连接并认证(从 session 文件恢复)
  51. // session 文件不存在时返回错误(不做交互式登录,session 需要预先生成)
  52. func (c *Client) Connect(ctx context.Context) error {
  53. storage := &session.FileStorage{Path: c.sessionPath}
  54. opts := telegram.Options{
  55. SessionStorage: storage,
  56. NoUpdates: true,
  57. Device: telegram.DeviceConfig{
  58. DeviceModel: c.account.Device,
  59. AppVersion: c.account.AppVersion,
  60. SystemVersion: c.account.SystemVersion,
  61. LangPack: c.account.LangPack,
  62. SystemLangCode: c.account.SystemLangCode,
  63. LangCode: c.account.LangCode,
  64. },
  65. }
  66. // Apply proxy if configured
  67. c.mu.Lock()
  68. proxyURL := c.proxyURL
  69. c.mu.Unlock()
  70. if proxyURL != "" {
  71. dialFunc, err := buildProxyDialer(proxyURL)
  72. if err != nil {
  73. log.Printf("[tg_client] failed to create proxy dialer: %v, connecting without proxy", err)
  74. } else {
  75. opts.Resolver = dcs.Plain(dcs.PlainOptions{Dial: dialFunc})
  76. log.Printf("[tg_client] connecting via proxy: %s", proxyURL)
  77. }
  78. }
  79. client := telegram.NewClient(c.account.AppID, c.account.AppHash, opts)
  80. runCtx, cancel := context.WithCancel(ctx)
  81. c.mu.Lock()
  82. c.tgc = client
  83. c.cancel = cancel
  84. c.ready = make(chan struct{})
  85. c.runErr = nil
  86. readyCh := c.ready
  87. c.mu.Unlock()
  88. errCh := make(chan error, 1)
  89. go func() {
  90. err := client.Run(runCtx, func(ctx context.Context) error {
  91. c.mu.Lock()
  92. c.api = client.API()
  93. close(readyCh)
  94. c.mu.Unlock()
  95. // Block until context is cancelled (Disconnect called)
  96. <-ctx.Done()
  97. return ctx.Err()
  98. })
  99. c.mu.Lock()
  100. c.runErr = err
  101. c.mu.Unlock()
  102. errCh <- err
  103. }()
  104. // Wait for ready or error
  105. select {
  106. case <-readyCh:
  107. return nil
  108. case err := <-errCh:
  109. if err != nil && err != context.Canceled {
  110. return err
  111. }
  112. return nil
  113. case <-ctx.Done():
  114. cancel()
  115. return ctx.Err()
  116. }
  117. }
  118. // Disconnect 断开连接
  119. func (c *Client) Disconnect() {
  120. c.mu.Lock()
  121. cancel := c.cancel
  122. c.mu.Unlock()
  123. if cancel != nil {
  124. cancel()
  125. }
  126. }
  127. // waitReady waits for the client to be connected and returns the api client
  128. func (c *Client) waitReady(ctx context.Context) (*tg.Client, error) {
  129. c.mu.Lock()
  130. readyCh := c.ready
  131. api := c.api
  132. c.mu.Unlock()
  133. if api != nil {
  134. return api, nil
  135. }
  136. select {
  137. case <-readyCh:
  138. c.mu.Lock()
  139. api = c.api
  140. c.mu.Unlock()
  141. return api, nil
  142. case <-ctx.Done():
  143. return nil, ctx.Err()
  144. }
  145. }
  146. // GetChannelInfo 获取频道/用户信息,通过用户名查找
  147. func (c *Client) GetChannelInfo(ctx context.Context, username string) (*ChannelInfo, error) {
  148. api, err := c.waitReady(ctx)
  149. if err != nil {
  150. return nil, err
  151. }
  152. username = strings.TrimPrefix(username, "@")
  153. resolved, err := api.ContactsResolveUsername(ctx, &tg.ContactsResolveUsernameRequest{
  154. Username: username,
  155. })
  156. if err != nil {
  157. return nil, wrapFloodWait(err)
  158. }
  159. info := &ChannelInfo{Username: username}
  160. // Look for channel/chat in the resolved chats
  161. for _, ch := range resolved.Chats {
  162. switch v := ch.(type) {
  163. case *tg.Channel:
  164. title := v.Title
  165. info.Title = title
  166. info.IsChannel = v.GetBroadcast()
  167. info.IsGroup = v.GetMegagroup()
  168. if count, ok := v.GetParticipantsCount(); ok {
  169. info.MemberCount = count
  170. }
  171. // Get full channel info for About
  172. accessHash, hasHash := v.GetAccessHash()
  173. if hasHash {
  174. full, ferr := api.ChannelsGetFullChannel(ctx, &tg.InputChannel{
  175. ChannelID: v.GetID(),
  176. AccessHash: accessHash,
  177. })
  178. if ferr == nil {
  179. if cf, ok := full.FullChat.(*tg.ChannelFull); ok {
  180. info.About = cf.GetAbout()
  181. if count, ok := cf.GetParticipantsCount(); ok && info.MemberCount == 0 {
  182. info.MemberCount = count
  183. }
  184. }
  185. }
  186. }
  187. return info, nil
  188. case *tg.Chat:
  189. info.Title = v.Title
  190. info.IsGroup = true
  191. info.MemberCount = v.ParticipantsCount
  192. return info, nil
  193. }
  194. }
  195. return info, nil
  196. }
  197. // GetMessages 获取频道历史消息
  198. // offsetID: 从哪条消息开始(断点续传)
  199. // limit: 最多取多少条
  200. // 返回的消息按 ID 从小到大排序
  201. func (c *Client) GetMessages(ctx context.Context, username string, offsetID, limit int) ([]Message, error) {
  202. api, err := c.waitReady(ctx)
  203. if err != nil {
  204. return nil, err
  205. }
  206. username = strings.TrimPrefix(username, "@")
  207. peer, err := c.resolveInputPeer(ctx, api, username)
  208. if err != nil {
  209. return nil, err
  210. }
  211. result, err := api.MessagesGetHistory(ctx, &tg.MessagesGetHistoryRequest{
  212. Peer: peer,
  213. OffsetID: offsetID,
  214. Limit: limit,
  215. })
  216. if err != nil {
  217. return nil, wrapFloodWait(err)
  218. }
  219. return extractMessages(result), nil
  220. }
  221. // GetPinnedMessages 获取置顶消息
  222. func (c *Client) GetPinnedMessages(ctx context.Context, username string) ([]Message, error) {
  223. api, err := c.waitReady(ctx)
  224. if err != nil {
  225. return nil, err
  226. }
  227. username = strings.TrimPrefix(username, "@")
  228. peer, err := c.resolveInputPeer(ctx, api, username)
  229. if err != nil {
  230. return nil, err
  231. }
  232. result, err := api.MessagesSearch(ctx, &tg.MessagesSearchRequest{
  233. Peer: peer,
  234. Filter: &tg.InputMessagesFilterPinned{},
  235. Limit: 100,
  236. })
  237. if err != nil {
  238. return nil, wrapFloodWait(err)
  239. }
  240. return extractMessages(result), nil
  241. }
  242. // VerifyUser 验证用户名是否存在,返回用户信息
  243. func (c *Client) VerifyUser(ctx context.Context, username string) (*UserInfo, error) {
  244. api, err := c.waitReady(ctx)
  245. if err != nil {
  246. return nil, err
  247. }
  248. username = strings.TrimPrefix(username, "@")
  249. resolved, err := api.ContactsResolveUsername(ctx, &tg.ContactsResolveUsernameRequest{
  250. Username: username,
  251. })
  252. if err != nil {
  253. if tgerr.Is(err, "USERNAME_NOT_OCCUPIED", "USERNAME_INVALID") {
  254. return &UserInfo{Username: username, Exists: false}, nil
  255. }
  256. return nil, wrapFloodWait(err)
  257. }
  258. // Check if the peer is a user
  259. if _, ok := resolved.Peer.(*tg.PeerUser); ok {
  260. for _, u := range resolved.Users {
  261. if user, ok := u.(*tg.User); ok {
  262. info := &UserInfo{
  263. ID: user.GetID(),
  264. Username: username,
  265. IsBot: user.GetBot(),
  266. IsPremium: user.GetPremium(),
  267. Exists: true,
  268. }
  269. if fn, ok := user.GetFirstName(); ok {
  270. info.FirstName = fn
  271. }
  272. if ln, ok := user.GetLastName(); ok {
  273. info.LastName = ln
  274. }
  275. if status, ok := user.GetStatus(); ok {
  276. if offline, ok := status.(*tg.UserStatusOffline); ok {
  277. t := time.Unix(int64(offline.GetWasOnline()), 0)
  278. info.LastOnline = &t
  279. }
  280. }
  281. return info, nil
  282. }
  283. }
  284. }
  285. // Check if it's a channel or group
  286. for _, ch := range resolved.Chats {
  287. switch v := ch.(type) {
  288. case *tg.Channel:
  289. return &UserInfo{
  290. ID: v.GetID(),
  291. Username: username,
  292. IsChannel: v.GetBroadcast(),
  293. IsGroup: v.GetMegagroup(),
  294. Exists: true,
  295. }, nil
  296. case *tg.Chat:
  297. return &UserInfo{
  298. ID: v.ID,
  299. IsGroup: true,
  300. Exists: true,
  301. }, nil
  302. }
  303. }
  304. return &UserInfo{Username: username, Exists: false}, nil
  305. }
  306. // GetGroupParticipants 获取群组/超级群组的成员列表(分页拉取全部)
  307. func (c *Client) GetGroupParticipants(ctx context.Context, username string) ([]GroupParticipant, error) {
  308. api, err := c.waitReady(ctx)
  309. if err != nil {
  310. return nil, err
  311. }
  312. username = strings.TrimPrefix(username, "@")
  313. // Resolve the channel/group
  314. resolved, err := api.ContactsResolveUsername(ctx, &tg.ContactsResolveUsernameRequest{
  315. Username: username,
  316. })
  317. if err != nil {
  318. return nil, wrapFloodWait(err)
  319. }
  320. // Find the channel in resolved chats
  321. var inputChannel *tg.InputChannel
  322. for _, ch := range resolved.Chats {
  323. switch v := ch.(type) {
  324. case *tg.Channel:
  325. accessHash, _ := v.GetAccessHash()
  326. inputChannel = &tg.InputChannel{
  327. ChannelID: v.GetID(),
  328. AccessHash: accessHash,
  329. }
  330. }
  331. }
  332. if inputChannel == nil {
  333. // Try as basic chat - get participants via MessagesGetFullChat
  334. if p, ok := resolved.Peer.(*tg.PeerChat); ok {
  335. return c.getChatParticipants(ctx, api, p.ChatID)
  336. }
  337. return nil, fmt.Errorf("无法解析群组: %s", username)
  338. }
  339. // Strategy: use ChannelParticipantsSearch with empty query (returns more than Recent),
  340. // then iterate alphabet queries to discover members beyond the 200 limit per query.
  341. seen := make(map[int64]bool)
  342. var allParticipants []GroupParticipant
  343. // Helper to extract users from a page
  344. extractUsers := func(cp *tg.ChannelsChannelParticipants) int {
  345. added := 0
  346. for _, u := range cp.Users {
  347. user, ok := u.(*tg.User)
  348. if !ok || seen[user.GetID()] {
  349. continue
  350. }
  351. seen[user.GetID()] = true
  352. p := GroupParticipant{
  353. ID: user.GetID(),
  354. IsBot: user.GetBot(),
  355. IsPremium: user.GetPremium(),
  356. }
  357. if un, ok := user.GetUsername(); ok {
  358. p.Username = un
  359. }
  360. if fn, ok := user.GetFirstName(); ok {
  361. p.FirstName = fn
  362. }
  363. if ln, ok := user.GetLastName(); ok {
  364. p.LastName = ln
  365. }
  366. allParticipants = append(allParticipants, p)
  367. added++
  368. }
  369. return added
  370. }
  371. // Phase 1: Search with empty query (gets up to ~200)
  372. totalCount := 0
  373. if err := c.fetchParticipantPages(ctx, api, inputChannel, "", seen, extractUsers, &totalCount); err != nil {
  374. if len(allParticipants) > 0 {
  375. return allParticipants, err
  376. }
  377. return nil, err
  378. }
  379. // Phase 2: If group has more members than we found, search by character sets to discover more.
  380. // We pace queries with jitter (2–4s) to avoid looking like a bot scanner and triggering FloodWait.
  381. // If FloodWait does hit, stop early and return what we already have — the calling task can
  382. // re-attempt later after the account cools down.
  383. if totalCount > len(allParticipants) && totalCount <= 10000 {
  384. queries := participantSearchQueries()
  385. for _, q := range queries {
  386. if ctx.Err() != nil {
  387. break
  388. }
  389. if len(allParticipants) >= totalCount {
  390. break // already collected everyone visible
  391. }
  392. beforeCount := len(allParticipants)
  393. err := c.fetchParticipantPages(ctx, api, inputChannel, q, seen, extractUsers, nil)
  394. if err != nil {
  395. if fwe, ok := err.(*FloodWaitError); ok {
  396. log.Printf("[tg_client] flood wait %ds during search q=%q for %s; returning %d/%d",
  397. fwe.Seconds, q, username, len(allParticipants), totalCount)
  398. } else {
  399. log.Printf("[tg_client] search q=%q for %s: %v (returning partial)", q, username, err)
  400. }
  401. break
  402. }
  403. if len(allParticipants) == beforeCount {
  404. continue // no new results; skip sleep and try next query
  405. }
  406. if err := jitterSleep(ctx, 2*time.Second, 4*time.Second); err != nil {
  407. return allParticipants, err
  408. }
  409. }
  410. }
  411. log.Printf("[tg_client] fetched %d/%d participants for %s", len(allParticipants), totalCount, username)
  412. return allParticipants, nil
  413. }
  414. // jitterSleep sleeps a random duration in [min, max) while respecting ctx.
  415. // Returns ctx.Err() if cancelled. Used to spread out TG API calls and avoid
  416. // looking like a deterministic scanner.
  417. func jitterSleep(ctx context.Context, min, max time.Duration) error {
  418. d := min + time.Duration(rand.Int64N(int64(max-min)))
  419. select {
  420. case <-ctx.Done():
  421. return ctx.Err()
  422. case <-time.After(d):
  423. return nil
  424. }
  425. }
  426. // participantSearchQueries returns search queries covering Latin, Cyrillic, Japanese,
  427. // Korean, and CJK scripts. TG's ChannelParticipantsSearch does substring matching on
  428. // first_name + last_name + username, so more starter-character coverage = more users
  429. // surfaced on groups beyond the 200-per-query cap. Total ~150 queries.
  430. func participantSearchQueries() []string {
  431. queries := make([]string, 0, 170)
  432. // Latin a-z
  433. for c := 'a'; c <= 'z'; c++ {
  434. queries = append(queries, string(c))
  435. }
  436. // Digits 0-9
  437. for c := '0'; c <= '9'; c++ {
  438. queries = append(queries, string(c))
  439. }
  440. // Cyrillic а-я
  441. for c := 'а'; c <= 'я'; c++ {
  442. queries = append(queries, string(c))
  443. }
  444. // Japanese Hiragana — common name-starter syllables
  445. queries = append(queries,
  446. "あ", "い", "う", "え", "お",
  447. "か", "さ", "た", "な", "ま",
  448. )
  449. // Korean Hangul — common initial syllables
  450. queries = append(queries,
  451. "가", "나", "다", "라", "마", "바", "사", "아", "자", "차",
  452. "카", "타", "파", "하",
  453. )
  454. // CJK: top Chinese surnames (百家姓 high frequency)
  455. surnames := []string{
  456. "王", "李", "张", "刘", "陈", "杨", "黄", "赵", "周", "吴",
  457. "徐", "孙", "马", "朱", "胡", "林", "何", "高", "郭", "罗",
  458. "谢", "宋", "唐", "许", "邓", "梁", "韩", "曹", "彭", "余",
  459. "潘", "袁", "蒋", "蔡", "卢", "田", "董", "叶", "程", "姜",
  460. }
  461. queries = append(queries, surnames...)
  462. // CJK: common given-name characters (高频二字名)
  463. given := []string{
  464. "伟", "芳", "娜", "秀", "敏", "静", "丽", "强", "磊", "军",
  465. "洋", "勇", "艳", "杰", "涛", "明", "超", "霞", "平", "刚",
  466. }
  467. queries = append(queries, given...)
  468. // CJK: common modifiers and city prefixes (covers nicknames/titles)
  469. misc := []string{
  470. "大", "小", "新", "老", "中", "天", "金", "一", "龙", "虎",
  471. "京", "沪", "深", "广", "杭", "苏",
  472. }
  473. queries = append(queries, misc...)
  474. return queries
  475. }
  476. // fetchParticipantPages paginates through ChannelParticipantsSearch results.
  477. func (c *Client) fetchParticipantPages(
  478. ctx context.Context,
  479. api *tg.Client,
  480. channel *tg.InputChannel,
  481. query string,
  482. seen map[int64]bool,
  483. extractUsers func(*tg.ChannelsChannelParticipants) int,
  484. outTotalCount *int,
  485. ) error {
  486. const pageSize = 200
  487. offset := 0
  488. for {
  489. result, err := api.ChannelsGetParticipants(ctx, &tg.ChannelsGetParticipantsRequest{
  490. Channel: channel,
  491. Filter: &tg.ChannelParticipantsSearch{Q: query},
  492. Offset: offset,
  493. Limit: pageSize,
  494. Hash: 0,
  495. })
  496. if err != nil {
  497. return wrapFloodWait(err)
  498. }
  499. cp, ok := result.(*tg.ChannelsChannelParticipants)
  500. if !ok || len(cp.Users) == 0 {
  501. break
  502. }
  503. if outTotalCount != nil && cp.Count > *outTotalCount {
  504. *outTotalCount = cp.Count
  505. }
  506. added := extractUsers(cp)
  507. offset += len(cp.Users)
  508. // If no new users were added in this page, stop
  509. if added == 0 || offset >= cp.Count {
  510. break
  511. }
  512. // Page interval: jittered to avoid a detectable request cadence.
  513. if err := jitterSleep(ctx, 800*time.Millisecond, 1500*time.Millisecond); err != nil {
  514. return err
  515. }
  516. }
  517. return nil
  518. }
  519. // getChatParticipants 获取普通群组的成员
  520. func (c *Client) getChatParticipants(ctx context.Context, api *tg.Client, chatID int64) ([]GroupParticipant, error) {
  521. full, err := api.MessagesGetFullChat(ctx, chatID)
  522. if err != nil {
  523. return nil, wrapFloodWait(err)
  524. }
  525. var participants []GroupParticipant
  526. for _, u := range full.Users {
  527. user, ok := u.(*tg.User)
  528. if !ok {
  529. continue
  530. }
  531. p := GroupParticipant{
  532. ID: user.GetID(),
  533. IsBot: user.GetBot(),
  534. IsPremium: user.GetPremium(),
  535. }
  536. if un, ok := user.GetUsername(); ok {
  537. p.Username = un
  538. }
  539. if fn, ok := user.GetFirstName(); ok {
  540. p.FirstName = fn
  541. }
  542. if ln, ok := user.GetLastName(); ok {
  543. p.LastName = ln
  544. }
  545. participants = append(participants, p)
  546. }
  547. return participants, nil
  548. }
  549. // buildProxyDialer creates a DialFunc that routes connections through the given proxy URL.
  550. // Supports socks5://, http://, https:// proxy protocols.
  551. func buildProxyDialer(rawURL string) (dcs.DialFunc, error) {
  552. u, err := url.Parse(rawURL)
  553. if err != nil {
  554. return nil, fmt.Errorf("parse proxy URL: %w", err)
  555. }
  556. switch u.Scheme {
  557. case "socks5", "socks5h":
  558. var auth *xproxy.Auth
  559. if u.User != nil {
  560. auth = &xproxy.Auth{User: u.User.Username()}
  561. if p, ok := u.User.Password(); ok {
  562. auth.Password = p
  563. }
  564. }
  565. dialer, err := xproxy.SOCKS5("tcp", u.Host, auth, xproxy.Direct)
  566. if err != nil {
  567. return nil, fmt.Errorf("create SOCKS5 dialer: %w", err)
  568. }
  569. ctxDialer, ok := dialer.(xproxy.ContextDialer)
  570. if !ok {
  571. return nil, fmt.Errorf("SOCKS5 dialer does not support DialContext")
  572. }
  573. return ctxDialer.DialContext, nil
  574. case "http", "https":
  575. // For HTTP proxies, use CONNECT tunneling
  576. return func(ctx context.Context, network, addr string) (net.Conn, error) {
  577. proxyConn, err := (&net.Dialer{Timeout: 15 * time.Second}).DialContext(ctx, "tcp", u.Host)
  578. if err != nil {
  579. return nil, fmt.Errorf("connect to proxy: %w", err)
  580. }
  581. // Set deadline for the CONNECT handshake
  582. if deadline, ok := ctx.Deadline(); ok {
  583. proxyConn.SetDeadline(deadline)
  584. } else {
  585. proxyConn.SetDeadline(time.Now().Add(15 * time.Second))
  586. }
  587. // Send CONNECT request
  588. connectReq := fmt.Sprintf("CONNECT %s HTTP/1.1\r\nHost: %s\r\n", addr, addr)
  589. if u.User != nil {
  590. pass, _ := u.User.Password()
  591. connectReq += fmt.Sprintf("Proxy-Authorization: Basic %s\r\n",
  592. encodeBasicAuth(u.User.Username(), pass))
  593. }
  594. connectReq += "\r\n"
  595. if _, err := proxyConn.Write([]byte(connectReq)); err != nil {
  596. proxyConn.Close()
  597. return nil, fmt.Errorf("write CONNECT: %w", err)
  598. }
  599. // Read HTTP response using bufio for proper line parsing
  600. br := bufio.NewReader(proxyConn)
  601. statusLine, err := br.ReadString('\n')
  602. if err != nil {
  603. proxyConn.Close()
  604. return nil, fmt.Errorf("read CONNECT status: %w", err)
  605. }
  606. // Parse status code from "HTTP/1.x NNN reason"
  607. parts := strings.SplitN(strings.TrimSpace(statusLine), " ", 3)
  608. if len(parts) < 2 || parts[1] != "200" {
  609. proxyConn.Close()
  610. return nil, fmt.Errorf("CONNECT failed: %s", strings.TrimSpace(statusLine))
  611. }
  612. // Consume remaining headers until empty line
  613. for {
  614. line, err := br.ReadString('\n')
  615. if err != nil {
  616. proxyConn.Close()
  617. return nil, fmt.Errorf("read CONNECT headers: %w", err)
  618. }
  619. if strings.TrimSpace(line) == "" {
  620. break
  621. }
  622. }
  623. // Clear deadline — gotd manages its own timeouts
  624. proxyConn.SetDeadline(time.Time{})
  625. return proxyConn, nil
  626. }, nil
  627. default:
  628. return nil, fmt.Errorf("unsupported proxy scheme: %s", u.Scheme)
  629. }
  630. }
  631. // encodeBasicAuth returns base64-encoded "user:password" for proxy auth.
  632. func encodeBasicAuth(user, password string) string {
  633. return base64.StdEncoding.EncodeToString([]byte(user + ":" + password))
  634. }
  635. // resolveInputPeer resolves a username to an InputPeer
  636. func (c *Client) resolveInputPeer(ctx context.Context, api *tg.Client, username string) (tg.InputPeerClass, error) {
  637. resolved, err := api.ContactsResolveUsername(ctx, &tg.ContactsResolveUsernameRequest{
  638. Username: username,
  639. })
  640. if err != nil {
  641. return nil, wrapFloodWait(err)
  642. }
  643. switch p := resolved.Peer.(type) {
  644. case *tg.PeerChannel:
  645. for _, ch := range resolved.Chats {
  646. if channel, ok := ch.(*tg.Channel); ok && channel.GetID() == p.ChannelID {
  647. accessHash, _ := channel.GetAccessHash()
  648. return &tg.InputPeerChannel{
  649. ChannelID: p.ChannelID,
  650. AccessHash: accessHash,
  651. }, nil
  652. }
  653. }
  654. return &tg.InputPeerChannel{ChannelID: p.ChannelID}, nil
  655. case *tg.PeerUser:
  656. for _, u := range resolved.Users {
  657. if user, ok := u.(*tg.User); ok && user.GetID() == p.UserID {
  658. accessHash, _ := user.GetAccessHash()
  659. return &tg.InputPeerUser{
  660. UserID: p.UserID,
  661. AccessHash: accessHash,
  662. }, nil
  663. }
  664. }
  665. return &tg.InputPeerUser{UserID: p.UserID}, nil
  666. case *tg.PeerChat:
  667. return &tg.InputPeerChat{ChatID: p.ChatID}, nil
  668. }
  669. return &tg.InputPeerEmpty{}, nil
  670. }
  671. // extractMessages extracts messages from a MessagesMessagesClass
  672. func extractMessages(result tg.MessagesMessagesClass) []Message {
  673. var rawMsgs []tg.MessageClass
  674. switch v := result.(type) {
  675. case *tg.MessagesMessages:
  676. rawMsgs = v.Messages
  677. case *tg.MessagesMessagesSlice:
  678. rawMsgs = v.Messages
  679. case *tg.MessagesChannelMessages:
  680. rawMsgs = v.Messages
  681. case *tg.MessagesMessagesNotModified:
  682. return nil
  683. }
  684. var msgs []Message
  685. for _, raw := range rawMsgs {
  686. switch m := raw.(type) {
  687. case *tg.Message:
  688. msg := Message{
  689. ID: m.GetID(),
  690. Text: m.GetMessage(),
  691. IsService: false,
  692. }
  693. // Extract forward source channel username
  694. if fwd, ok := m.GetFwdFrom(); ok {
  695. if fromID, ok := fwd.GetFromID(); ok {
  696. if peerCh, ok := fromID.(*tg.PeerChannel); ok {
  697. _ = peerCh // We'd need channel map to resolve username; skip for now
  698. }
  699. }
  700. }
  701. // Extract t.me links from text
  702. msg.Links = tmeRegexp.FindAllString(msg.Text, -1)
  703. msgs = append(msgs, msg)
  704. case *tg.MessageService:
  705. msgs = append(msgs, Message{
  706. ID: m.GetID(),
  707. IsService: true,
  708. })
  709. }
  710. }
  711. // Sort by ID ascending
  712. sort.Slice(msgs, func(i, j int) bool {
  713. return msgs[i].ID < msgs[j].ID
  714. })
  715. return msgs
  716. }
  717. // isFloodWait 检查错误是否是 FloodWait,提取等待时间
  718. func isFloodWait(err error) (bool, int) {
  719. if d, ok := tgerr.AsFloodWait(err); ok {
  720. return true, int(d.Seconds())
  721. }
  722. return false, 0
  723. }
  724. // wrapFloodWait wraps a FloodWait error into FloodWaitError
  725. func wrapFloodWait(err error) error {
  726. if ok, secs := isFloodWait(err); ok {
  727. return &FloodWaitError{Seconds: secs}
  728. }
  729. return err
  730. }