root/tgex-backend
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ручное парсер

Browse Source
This commit is contained in:
Artem Tsyrulnikov
2026-02-04 13:38:03 +03:00
parent 09145e510a
commit 6fc6a60bce
10 changed files with 145 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
tg_parser/internal/adapter/database/get_channels_with_tracked_posts.go
View File

@@ -18,12 +18,14 @@ func (d *Database) GetChannelsWithTrackedPosts(ctx context.Context) ([]domain.Ch
c.title,
c.access_hash,
c.pts,
c.invite_link
c.invite_link,
c.is_accessible
FROM channel c
INNER JOIN placement p ON p.channel_id = c.id
INNER JOIN placement_post pp ON pp.placement_id = p.id
INNER JOIN post po ON po.id = pp.post_id
WHERE po.deleted_from_channel_at IS NULL;`
WHERE po.deleted_from_channel_at IS NULL
AND c.is_accessible = true;`
txOrPool := transaction.TryExtractTX(ctx)
@@ -50,13 +52,14 @@ WHERE po.deleted_from_channel_at IS NULL;`
}
type getChannelsWithTrackedPostsDTO struct {
ID pgtype.UUID
TelegramID pgtype.Int8
Username pgtype.Text
Title pgtype.Text
AccessHash pgtype.Int8
Pts pgtype.Int4
InviteLink pgtype.Text
ID pgtype.UUID
TelegramID pgtype.Int8
Username pgtype.Text
Title pgtype.Text
AccessHash pgtype.Int8
Pts pgtype.Int4
InviteLink pgtype.Text
IsAccessible pgtype.Bool
}
func (dto *getChannelsWithTrackedPostsDTO) destination() []any {
@@ -68,17 +71,19 @@ func (dto *getChannelsWithTrackedPostsDTO) destination() []any {
&dto.AccessHash,
&dto.Pts,
&dto.InviteLink,
&dto.IsAccessible,
}
}
func (dto *getChannelsWithTrackedPostsDTO) toDomain() domain.Channel {
return domain.Channel{
ID: dto.ID.Bytes,
TelegramID: domain.NormalizeChatID(dto.TelegramID.Int64),
Username: dto.Username.String,
Title: dto.Title.String,
AccessHash: dto.AccessHash.Int64,
Pts: int(dto.Pts.Int32),
InviteLink: dto.InviteLink.String,
ID: dto.ID.Bytes,
TelegramID: domain.NormalizeChatID(dto.TelegramID.Int64),
Username: dto.Username.String,
Title: dto.Title.String,
AccessHash: dto.AccessHash.Int64,
Pts: int(dto.Pts.Int32),
InviteLink: dto.InviteLink.String,
IsAccessible: dto.IsAccessible.Bool,
}
}

13
tg_parser/internal/adapter/database/update_channel.go
View File

@@ -5,11 +5,9 @@ import (
"fmt"
"strings"
"github.com/jackc/pgx/v5/pgtype"
"github.com/rs/zerolog/log"
"github.com/TelegramExchange/pkg/transaction"
"github.com/TelegramExchange/tgex-backend/tg_parser/internal/domain"
"github.com/jackc/pgx/v5/pgtype"
)
func (d *Database) UpdateChannel(ctx context.Context, channel domain.Channel) error {
@@ -42,15 +40,6 @@ WHERE id = $1;`
inviteLinkDTO = pgtype.Text{Valid: false}
}
// Log warning if both identity fields are empty (will violate CHECK constraint)
if !usernameDTO.Valid && !inviteLinkDTO.Valid {
log.Warn().
Str("channel_id", channel.ID.String()).
Int64("telegram_id", channel.TelegramID).
Str("title", channel.Title).
Msg("UpdateChannel: both username and invite_link are empty, this may violate CHECK constraint")
}
dto := updateChannelDTO{
ID: pgtype.UUID{Bytes: channel.ID, Valid: true},
TelegramID: pgtype.Int8{Int64: channel.TelegramID, Valid: true},

6
tg_parser/internal/adapter/telegram/history.go
View File

@@ -31,7 +31,7 @@ func (t *Telegram) GetChannelHistory(ctx context.Context, channel domain.Channel
return nil, nil
}
placements := make([]domain.Post, 0, len(messages.Messages))
posts := make([]domain.Post, 0, len(messages.Messages))
for _, raw := range messages.Messages {
m, ok := raw.(*tg.Message)
@@ -42,8 +42,8 @@ func (t *Telegram) GetChannelHistory(ctx context.Context, channel domain.Channel
text := messageToHTML(m.Message, m.Entities)
publishedAt := time.Unix(int64(m.Date), 0).UTC()
p := domain.NewPost(channel, m.ID, text, m.Views, publishedAt)
placements = append(placements, p)
posts = append(posts, p)
}
return placements, nil
return posts, nil
}

2
tg_parser/internal/adapter/telegram/resolve.go
View File

@@ -5,7 +5,6 @@ import (
"fmt"
"github.com/TelegramExchange/tgex-backend/tg_parser/internal/domain"
"github.com/google/uuid"
"github.com/gotd/td/tg"
)
@@ -49,7 +48,6 @@ func (t *Telegram) ParseChannelMeta(ctx context.Context, username string) (domai
}
return domain.Channel{
ID: uuid.New(),
TelegramID: domain.ChatIDFromChannelID(ch.ID),
Username: username,
Title: ch.Title,

3
tg_parser/internal/adapter/telegram/resolve_invite.go
View File

@@ -5,7 +5,6 @@ import (
"fmt"
"github.com/TelegramExchange/tgex-backend/tg_parser/internal/domain"
"github.com/google/uuid"
"github.com/gotd/td/telegram/deeplink"
"github.com/gotd/td/tg"
)
@@ -64,12 +63,12 @@ func (t *Telegram) ParseChannelMetaByInvite(ctx context.Context, inviteLink stri
}
return domain.Channel{
ID: uuid.New(),
TelegramID: domain.ChatIDFromChannelID(channel.ID),
Username: username,
Title: channel.Title,
AccessHash: channel.AccessHash,
Pts: pts,
InviteLink: inviteLink,
IsAccessible: true,
}, nil
}

13
tg_parser/internal/usecase/fetch_channel_meta.go
View File

@@ -2,10 +2,21 @@ package usecase
import (
"context"
"fmt"
"github.com/TelegramExchange/tgex-backend/tg_parser/internal/domain"
)
func (uc *UseCase) FetchChannelMeta(ctx context.Context, username string) (domain.Channel, error) {
return uc.telegram.ParseChannelMeta(ctx, username)
channel, err := uc.telegram.ParseChannelMeta(ctx, username)
if err != nil {
return domain.Channel{}, fmt.Errorf("uc.telegram.ParseChannelMeta: %s", err)
}
err = uc.database.UpdateChannelIfNotAccessible(ctx, channel.TelegramID)
if err != nil {
return domain.Channel{}, fmt.Errorf("uc.database.UpdateChannelIfNotAccessible: %s", err)
}
return channel, nil
}

13
tg_parser/internal/usecase/fetch_channel_meta_by_invite.go
View File

@@ -2,10 +2,21 @@ package usecase
import (
"context"
"fmt"
"github.com/TelegramExchange/tgex-backend/tg_parser/internal/domain"
)
func (uc *UseCase) FetchChannelMetaByInvite(ctx context.Context, inviteLink string) (domain.Channel, error) {
return uc.telegram.ParseChannelMetaByInvite(ctx, inviteLink)
channel, err := uc.telegram.ParseChannelMetaByInvite(ctx, inviteLink)
if err != nil {
return domain.Channel{}, fmt.Errorf("uc.telegram.ParseChannelMetaByInvite: %s", err)
}
err = uc.database.UpdateChannelIfNotAccessible(ctx, channel.TelegramID)
if err != nil {
return domain.Channel{}, fmt.Errorf("uc.database.UpdateChannelIfNotAccessible: %s", err)
}
return channel, nil
}

157
tg_parser/internal/usecase/fetch_channels.go
View File

@@ -2,6 +2,7 @@ package usecase
import (
"context"
"errors"
"fmt"
"github.com/TelegramExchange/pkg/transaction"
@@ -16,21 +17,18 @@ func (uc *UseCase) FetchChannels(ctx context.Context) error {
for _, c := range channels {
if c.Pts == 0 {
updatedChannel, err := uc.initializeChannel(ctx, c)
if err != nil {
return fmt.Errorf("uc.initializeChannel: %w", err)
}
c = updatedChannel
}
if c.Pts == 0 {
log.Warn().Int64("telegram_id", c.TelegramID).Msg("Channel pts still empty, skipping diff")
go uc.initializeChannel(ctx, c)
continue
}
err := uc.processChannel(ctx, c)
if err != nil {
if isPrivateChannelError(err) {
log.Warn().Int64("telegram_id", c.TelegramID).Msg("Private channel, attempting rejoin")
go uc.rejoinChannel(ctx, c)
return nil
}
return fmt.Errorf("uc.processChannel: %w", err)
}
}
@@ -38,37 +36,13 @@ func (uc *UseCase) FetchChannels(ctx context.Context) error {
return nil
}
func isPrivateChannelError(err error) bool {
return tgerr.Is(err, "CHANNEL_PRIVATE", "CHANNEL_INVALID", "CHANNEL_FORBIDDEN")
}
func (uc *UseCase) processChannel(ctx context.Context, channel domain.Channel) error {
diff, err := uc.telegram.GetChannelDiff(ctx, channel, 20)
if err != nil {
if isPrivateChannelError(err) {
log.Warn().Err(err).Int64("telegram_id", channel.TelegramID).Msg("Private channel, attempting rejoin")
if channel.InviteLink == "" {
log.Warn().Int64("telegram_id", channel.TelegramID).Msg("No invite link stored, marking inaccessible")
channel.IsAccessible = false
if err := uc.database.UpdateChannel(ctx, channel); err != nil {
return fmt.Errorf("database.UpdateChannel (mark inaccessible): %w", err)
}
return nil
}
updatedChannel, joinErr := uc.telegram.ParseChannelMetaByInvite(ctx, channel.InviteLink)
if joinErr != nil {
log.Warn().Err(joinErr).Int64("telegram_id", channel.TelegramID).Msg("Invite rejoin failed, marking inaccessible")
channel.IsAccessible = false
if err := uc.database.UpdateChannel(ctx, channel); err != nil {
return fmt.Errorf("database.UpdateChannel (mark inaccessible): %w", err)
}
return nil
}
updatedChannel.ID = channel.ID
updatedChannel.InviteLink = channel.InviteLink
updatedChannel.IsAccessible = true
updatedChannel.Pts = 0
if err := uc.database.UpdateChannel(ctx, updatedChannel); err != nil {
return fmt.Errorf("database.UpdateChannel: %w", err)
}
return nil
}
return fmt.Errorf("telegram.GetChannelDiff: %w", err)
}
@@ -97,49 +71,90 @@ func (uc *UseCase) processChannel(ctx context.Context, channel domain.Channel) e
if err != nil {
return fmt.Errorf("database.UpdateChannel: %w", err)
}
} else {
channel.Pts = diff.NewPts
err = uc.database.UpdateChannel(ctx, channel)
if err != nil {
return fmt.Errorf("database.UpdateChannel: %w", err)
}
return nil
}
channel.Pts = diff.NewPts
err = uc.database.UpdateChannel(ctx, channel)
if err != nil {
return fmt.Errorf("database.UpdateChannel: %w", err)
}
return nil
}
func isPrivateChannelError(err error) bool {
return tgerr.Is(err, "CHANNEL_PRIVATE", "CHANNEL_INVALID", "CHANNEL_FORBIDDEN")
}
func (uc *UseCase) rejoinChannel(ctx context.Context, channel domain.Channel) {
defer func() {
err := uc.database.UpdateChannel(ctx, channel)
if err != nil {
log.Err(err).Msg("database.UpdateChannel (rejoinChannel)")
}
}()
func (uc *UseCase) initializeChannel(ctx context.Context, channel domain.Channel) (domain.Channel, error) {
c := channel
if channel.Username != "" {
updated, err := uc.telegram.ParseChannelMeta(ctx, channel.Username)
if err != nil {
return domain.Channel{}, fmt.Errorf("telegram.ParseChannelMeta: %w", err)
}
c = updated
} else if channel.AccessHash == 0 {
return domain.Channel{}, fmt.Errorf("channel %s missing access hash", channel.ID)
} else {
pts, err := uc.telegram.GetChannelPTS(ctx, channel)
if err != nil {
return domain.Channel{}, fmt.Errorf("telegram.GetChannelPTS: %w", err)
}
c.Pts = pts
if channel.InviteLink == "" {
log.Warn().Int64("telegram_id", channel.TelegramID).Msg("No invite link stored, marking inaccessible")
channel.IsAccessible = false
return
}
// Preserve persistent channel ID from the database so foreign keys remain valid.
c.ID = channel.ID
posts, err := uc.telegram.GetChannelHistory(ctx, c, 20)
updated, err := uc.telegram.ParseChannelMetaByInvite(ctx, channel.InviteLink)
if err != nil {
return domain.Channel{}, fmt.Errorf("telegram.GetChannelHistory: %w", err)
log.Warn().Err(err).Int64("telegram_id", channel.TelegramID).Msg("Invite rejoin failed, marking inaccessible")
channel.IsAccessible = false
return
}
channel.TelegramID = updated.TelegramID
channel.Title = updated.Title
channel.AccessHash = 0
channel.Pts = 0
channel.IsAccessible = true
}
func (uc *UseCase) initializeChannel(ctx context.Context, channel domain.Channel) {
var (
updated domain.Channel
err error
)
switch {
case channel.Username != "":
updated, err = uc.telegram.ParseChannelMeta(ctx, channel.Username)
case channel.InviteLink != "":
updated, err = uc.telegram.ParseChannelMetaByInvite(ctx, channel.InviteLink)
default:
err = errors.New("channel state invalid")
}
if err != nil {
if isPrivateChannelError(err) {
channel.IsAccessible = false
err = uc.database.UpdateChannel(ctx, channel)
if err != nil {
log.Err(err).Msg("initializeChannel.uc.database.UpdateChannel")
}
}
log.Error().Err(err).Msg("initializeChannel.ParseChannel")
return
}
channel.Title = updated.Title
channel.AccessHash = updated.AccessHash
channel.Pts = updated.Pts
const initPosts = 20
posts, err := uc.telegram.GetChannelHistory(ctx, channel, initPosts)
if err != nil {
log.Error().Err(err).Msg("telegram.GetChannelHistory")
return
}
err = transaction.Wrap(ctx, func(ctx context.Context) error {
err = uc.database.UpdateChannel(ctx, c)
err = uc.database.UpdateChannel(ctx, channel)
if err != nil {
return fmt.Errorf("database.UpdateChannel: %w", err)
}
@@ -155,8 +170,6 @@ func (uc *UseCase) initializeChannel(ctx context.Context, channel domain.Channel
return nil
})
if err != nil {
return domain.Channel{}, fmt.Errorf("transaction.Wrap: %w", err)
log.Error().Err(err).Msg("transaction.Wrap")
}
return c, nil
}

8
tg_parser/internal/usecase/fetch_views.go
View File

@@ -23,6 +23,14 @@ func (uc *UseCase) FetchViews(ctx context.Context) error {
err = uc.telegram.UpdatePostsViews(ctx, channel, posts)
if err != nil {
if isPrivateChannelError(err) {
channel.IsAccessible = false
err = uc.database.UpdateChannel(ctx, channel)
if err != nil {
log.Err(err).Msg("initializeChannel.uc.database.UpdateChannel")
}
return nil
}
return err
}

1
tg_parser/internal/usecase/usecase.go
View File

@@ -20,6 +20,7 @@ type Database interface {
CreateViewsSnapshot(ctx context.Context, snapshot domain.ViewsSnapshot) error
UpdateChannelIfNotAccessible(ctx context.Context, channel domain.Channel) error
GetChannels(ctx context.Context) []domain.Channel
UpdateChannel(ctx context.Context, channel domain.Channel) error
GetChannelsWithTrackedPosts(ctx context.Context) ([]domain.Channel, error)