red-future/video-factory
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1

Browse Source
This commit is contained in:
admin
2026-08-04 15:25:11 +08:00
parent 2e9e349c9a
commit e8fa311b59
28 changed files with 972 additions and 402 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile
+15 -2
View File
@@ -1,3 +1,13 @@
# ==================== 前端构建（ui-src） ====================
FROM node:20-alpine AS ui-builder
RUN apk add --no-cache git
WORKDIR /build-ui
COPY video-factory/ui-src/package.json video-factory/ui-src/package-lock.json ./
RUN npm ci --registry=https://registry.npmmirror.com
COPY video-factory/ui-src/ ./
RUN npm run build
# ==================== 后端构建 ====================
FROM golang:alpine AS builder
RUN sed -i 's/dl-cdn.alpinelinux.org/mirrors.aliyun.com/g' /etc/apk/repositories \
&& apk add --no-cache git ca-certificates tzdata
@@ -7,10 +17,11 @@ ENV GOPROXY=https://goproxy.cn,direct
ENV CGO_ENABLED=0
ENV GOTOOLCHAIN=auto
WORKDIR /build
COPY . .
COPY video-factory/ .
RUN go mod download && go mod tidy
RUN go build -ldflags="-s -w" -o main ./main.go
# ==================== 运行镜像 ====================
FROM alpine:3.19
RUN sed -i 's/dl-cdn.alpinelinux.org/mirrors.aliyun.com/g' /etc/apk/repositories \
&& apk add --no-cache ca-certificates tzdata ffmpeg
@@ -22,6 +33,8 @@ COPY --from=builder /build/prompt.md .
COPY --from=builder /build/negative_prompt.md .
COPY --from=builder /build/default_first_frame.png .
COPY --from=builder /build/main .
RUN printf '#!/bin/sh\nif [ -d /app/short_drama.db ]; then rm -rf /app/short_drama.db; fi\ntouch /app/short_drama.db 2>/dev/null || true\nexec ./main\n' > /app/entrypoint.sh && chmod +x /app/entrypoint.sh
COPY --from=ui-builder /build-ui/dist ./ui-src/dist
RUN mkdir -p /app/data \
&& printf '#!/bin/sh\nfor db in business.db system.db finance.db; do\n if [ -d /app/data/$db ]; then rm -rf /app/data/$db; fi\n touch /app/data/$db 2>/dev/null || true\ndone\nexec ./main\n' > /app/entrypoint.sh && chmod +x /app/entrypoint.sh
EXPOSE 3006
ENTRYPOINT ["/app/entrypoint.sh"]
common/auth_middleware.go
+8
View File
@@ -28,6 +28,14 @@ func Auth(r *ghttp.Request) {
return
}
// 前端静态资源放行（合并部署后 UI 由后端服务，浏览器请求不带 Authorization）：
// 仅放行 /（入口页）与 /assets/*（构建产物），hash 路由下 SPA 页面只会请求这两个路径；
// 其余路径（包括与 SPA 页面同名的 /drama、/customer 等）一律走 API 鉴权，无鉴权绕过
if r.Method == http.MethodGet && (path == "/" || strings.HasPrefix(path, "/assets/")) {
r.Middleware.Next()
return
}
auth := r.Header.Get("Authorization")
if auth == "" || !strings.HasPrefix(auth, "Bearer ") {
r.Response.WriteJson(ghttp.DefaultHandlerResponse{
config.yml
+9 -1
View File
@@ -1,8 +1,16 @@
database:
default:
name: short_drama.db
name: data/business.db
type: sqlite
debug: true # (可选)开启调试模式
system:
name: data/system.db
type: sqlite
debug: true
finance:
name: data/finance.db
type: sqlite
debug: true
cache:
ttl: 60 # DAO查询缓存时间(秒)，0为禁用缓存
server:
main.go
+27
View File
@@ -92,6 +92,33 @@ func main() {
r.Response.ServeFile(filePath)
})
// ==================== 前端静态资源服务（SPA，前后端合并部署） ====================
// 静态资源目录为 ui-src/dist：本地是 npm run build 产物，Docker 运行镜像由
// ui-builder 阶段将 dist 注入 /app/ui-src/dist，本地与容器路径一致，无需拷贝。
// 不用 AddStaticPath：GF 静态前缀有边界检查（前缀后一字符必须为 /），
// "/" 前缀无法命中 /assets/* 子路径，只能服务 "/" 本身。
// 前端使用 hash 路由，浏览器只请求 / 与 /assets/*，BindHandler("/*") 精确覆盖即可，
// 无需 SPA fallback；具体 API 路由与 /workspace/* 优先级更高不受影响，
// 且 auth 中间件仅放行 GET / 与 GET /assets/*，其余路径一律走鉴权
if st, err := os.Stat("ui-src/dist"); err == nil && st.IsDir() {
commonHttp.Httpserver.BindHandler("/*", func(r *ghttp.Request) {
path := strings.TrimPrefix(r.URL.Path, "/")
if strings.Contains(path, "..") {
r.Response.WriteStatus(http.StatusNotFound)
return
}
if path == "" || path == "index.html" {
path = "index.html"
}
filePath := filepath.Join("ui-src/dist", path)
if st, err := os.Stat(filePath); err == nil && !st.IsDir() {
r.Response.ServeFile(filePath)
return
}
r.Response.WriteStatus(http.StatusNotFound)
})
}
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
short_drama.db
BIN
View File
Binary file not shown.
shortdrama/consts/table_name.go
+7
View File
@@ -19,3 +19,10 @@ const (
TableNameRegionPricing = "region_pricing"
TableNameUserModelConfig = "user_model_config"
)
// 数据库组：默认组(default)=business.db、system=system.db、finance=finance.db
const (
DbGroupBusiness = "" // 默认组，对应 business.db
DbGroupSystem = "system"
DbGroupFinance = "finance"
)
shortdrama/controller/payment_order_controller.go
+12 -4
View File
@@ -64,8 +64,12 @@ func (c *paymentOrder) ConfirmOffline(ctx context.Context, req *dto.ConfirmOffli
func (c *paymentOrder) NotifyWechat(ctx context.Context, req *struct{}) (res *struct{}, err error) {
r := g.RequestFromCtx(ctx)
body := r.GetBody()
_ = service.PaymentService.HandleNotify(ctx, "wechat", body)
r.Response.WriteString("<xml><return_code><![CDATA[SUCCESS]]></return_code><return_msg><![CDATA[OK]]></return_msg></xml>")
if err := service.PaymentService.HandleNotify(ctx, "wechat", body); err != nil {
g.Log().Errorf(ctx, "wechat payment notify handle failed: %v", err)
r.Response.WriteString("<xml><return_code><![CDATA[FAIL]]></return_code><return_msg><![CDATA[FAIL]]></return_msg></xml>")
} else {
r.Response.WriteString("<xml><return_code><![CDATA[SUCCESS]]></return_code><return_msg><![CDATA[OK]]></return_msg></xml>")
}
r.Exit()
return nil, nil
}
@@ -73,8 +77,12 @@ func (c *paymentOrder) NotifyWechat(ctx context.Context, req *struct{}) (res *st
func (c *paymentOrder) NotifyAlipay(ctx context.Context, req *struct{}) (res *struct{}, err error) {
r := g.RequestFromCtx(ctx)
body := r.GetBody()
_ = service.PaymentService.HandleNotify(ctx, "alipay", body)
r.Response.WriteString("success")
if err := service.PaymentService.HandleNotify(ctx, "alipay", body); err != nil {
g.Log().Errorf(ctx, "alipay payment notify handle failed: %v", err)
r.Response.WriteString("failure")
} else {
r.Response.WriteString("success")
}
r.Exit()
return nil, nil
}
shortdrama/dao/account_transaction_dao.go
+7 -7
View File
@@ -16,7 +16,7 @@ type accountTransactionDao struct{}
func init() {
ctx := context.Background()
_, err := g.DB().Exec(ctx, `CREATE TABLE IF NOT EXISTS `+consts.TableNameAccountTransaction+` (
_, err := g.DB(consts.DbGroupFinance).Exec(ctx, `CREATE TABLE IF NOT EXISTS `+consts.TableNameAccountTransaction+` (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL DEFAULT 0,
type TEXT NOT NULL DEFAULT '',
@@ -30,10 +30,10 @@ func init() {
if err != nil {
g.Log().Warningf(ctx, "create account_transaction table failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "CREATE INDEX IF NOT EXISTS idx_at_user ON "+consts.TableNameAccountTransaction+"(user_id)"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "CREATE INDEX IF NOT EXISTS idx_at_user ON "+consts.TableNameAccountTransaction+"(user_id)"); err != nil {
g.Log().Warningf(ctx, "create index idx_at_user failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "ALTER TABLE "+consts.TableNameAccountTransaction+" ADD COLUMN order_no TEXT NOT NULL DEFAULT ''"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "ALTER TABLE "+consts.TableNameAccountTransaction+" ADD COLUMN order_no TEXT NOT NULL DEFAULT ''"); err != nil {
g.Log().Warningf(ctx, "add column order_no failed: %v", err)
}
}
@@ -43,9 +43,9 @@ func clearAccountTransactionCache(ctx context.Context, userId int64) {
}
func (d *accountTransactionDao) Insert(ctx context.Context, data *entity.AccountTransaction) (int64, error) {
r, err := g.DB().Exec(ctx,
"INSERT INTO "+consts.TableNameAccountTransaction+" (user_id, type, amount, balance_before, balance_after, remark, created_by, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, datetime('now','localtime'))",
data.UserId, data.Type, data.Amount, data.BalanceBefore, data.BalanceAfter, data.Remark, data.CreatedBy)
r, err := g.DB(consts.DbGroupFinance).Exec(ctx,
"INSERT INTO "+consts.TableNameAccountTransaction+" (user_id, type, amount, balance_before, balance_after, remark, created_by, order_no, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, datetime('now','localtime'))",
data.UserId, data.Type, data.Amount, data.BalanceBefore, data.BalanceAfter, data.Remark, data.CreatedBy, data.OrderNo)
if err != nil {
return 0, err
}
@@ -54,7 +54,7 @@ func (d *accountTransactionDao) Insert(ctx context.Context, data *entity.Account
}
func (d *accountTransactionDao) ListByUser(ctx context.Context, userId int64, typeFilter string, page, pageSize int) ([]*entity.AccountTransaction, int, error) {
m := g.DB().Model(consts.TableNameAccountTransaction).Ctx(ctx).Where("user_id", userId)
m := g.DB(consts.DbGroupFinance).Model(consts.TableNameAccountTransaction).Ctx(ctx).Where("user_id", userId)
if typeFilter != "" {
m = m.Where("type", typeFilter)
}
shortdrama/dao/agent_profile_dao.go
+94 -37
View File
@@ -31,51 +31,108 @@ type agentListRow struct {
}
// ListAgentWithProfile 按条件分页查询代理商列表（含 profile 信息）
// 拆库后改为多个单表查询 + 内存合并，避免跨表 JOIN
func (d *agentProfileDao) ListAgentWithProfile(ctx context.Context, keyword, phone, province, region, expiredAtFrom, expiredAtTo string, page, pageSize int) ([]*agentListRow, int, error) {
m := g.DB().Model(consts.TableNameUser+" u").
LeftJoin(consts.TableNameAgentProfile+" ap", "ap.user_id = u.id").
Where("u.role", "agent")
// 分页+筛选查询不缓存（参数不同导致缓存键碰撞）
if keyword != "" {
m = m.Where("u.name LIKE ? OR u.username LIKE ? OR u.phone LIKE ?", "%"+keyword+"%", "%"+keyword+"%", "%"+keyword+"%")
}
if phone != "" {
m = m.Where("u.phone LIKE ?", "%"+phone+"%")
}
if province != "" {
m = m.Where("u.province = ?", province)
}
if region != "" {
m = m.Where("u.region = ?", region)
}
if expiredAtFrom != "" {
m = m.Where("ap.expired_at >= ?", expiredAtFrom)
}
if expiredAtTo != "" {
m = m.Where("ap.expired_at <= ?", expiredAtTo)
}
if page < 1 {
page = 1
}
if pageSize < 1 {
pageSize = 20
}
m = m.Fields(
"u.id", "u.username", "u.phone", "u.name", "u.province", "u.region",
"ap.expired_at", "ap.region_protected", "ap.max_customers",
)
r, total, err := m.OrderAsc("u.id").Limit(pageSize).Offset((page - 1) * pageSize).AllAndCount(false)
// 有效期筛选：先单表查出符合条件的代理商 user_id
var agentIds []int64
if expiredAtFrom != "" || expiredAtTo != "" {
m := g.DB(consts.DbGroupFinance).Model(consts.TableNameAgentProfile).Ctx(ctx).Fields("user_id")
if expiredAtFrom != "" {
m = m.Where("expired_at >= ?", expiredAtFrom)
}
if expiredAtTo != "" {
m = m.Where("expired_at <= ?", expiredAtTo)
}
r, err := m.All()
if err != nil {
return nil, 0, err
}
for _, row := range r {
agentIds = append(agentIds, row["user_id"].Int64())
}
if len(agentIds) == 0 {
return make([]*agentListRow, 0), 0, nil
}
}
// 代理商主查询（user 表，role=agent + 关键字/手机号/省市筛选）
m := g.DB(consts.DbGroupFinance).Model(consts.TableNameUser).Ctx(ctx).Where("role", "agent")
if keyword != "" {
m = m.Where("name LIKE ? OR username LIKE ? OR phone LIKE ?", "%"+keyword+"%", "%"+keyword+"%", "%"+keyword+"%")
}
if phone != "" {
m = m.Where("phone LIKE ?", "%"+phone+"%")
}
if province != "" {
m = m.Where("province = ?", province)
}
if region != "" {
m = m.Where("region = ?", region)
}
if len(agentIds) > 0 {
m = m.WhereIn("id", agentIds)
}
total, err := m.Count()
if err != nil {
return nil, 0, err
}
var rows []*agentListRow
err = r.Structs(&rows)
return rows, total, err
var users []*entity.User
err = m.OrderAsc("id").Limit(pageSize).Offset((page - 1) * pageSize).Scan(&users)
if err != nil {
return nil, 0, err
}
if len(users) == 0 {
return make([]*agentListRow, 0), total, nil
}
// 批量补 profile 信息
pageIds := make([]int64, 0, len(users))
for _, u := range users {
pageIds = append(pageIds, u.Id)
}
profiles := make(map[int64]*entity.AgentProfile)
var profileList []*entity.AgentProfile
if err := g.DB(consts.DbGroupFinance).Model(consts.TableNameAgentProfile).Ctx(ctx).
WhereIn("user_id", pageIds).Scan(&profileList); err != nil {
return nil, 0, err
}
for _, p := range profileList {
profiles[p.UserId] = p
}
// 内存合并组装结果（无 profile 时输出零值，与 LEFT JOIN 语义一致）
rows := make([]*agentListRow, 0, len(users))
for _, u := range users {
row := &agentListRow{
Id: u.Id,
Username: u.Username,
Phone: u.Phone,
Name: u.Name,
Province: u.Province,
Region: u.Region,
}
if p, ok := profiles[u.Id]; ok {
row.ExpiredAt = p.ExpiredAt
if p.RegionProtected {
row.RegionProtected = 1
}
row.MaxCustomers = p.MaxCustomers
}
rows = append(rows, row)
}
return rows, total, nil
}
func init() {
ctx := context.Background()
_, err := g.DB().Exec(ctx, `CREATE TABLE IF NOT EXISTS `+consts.TableNameAgentProfile+` (
_, err := g.DB(consts.DbGroupFinance).Exec(ctx, `CREATE TABLE IF NOT EXISTS `+consts.TableNameAgentProfile+` (
user_id INTEGER PRIMARY KEY,
max_customers INTEGER NOT NULL DEFAULT 0,
renewals INTEGER NOT NULL DEFAULT 0,
@@ -85,13 +142,13 @@ func init() {
if err != nil {
g.Log().Warningf(ctx, "create agent_profile table failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "ALTER TABLE "+consts.TableNameAgentProfile+" ADD COLUMN expired_at DATETIME"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "ALTER TABLE "+consts.TableNameAgentProfile+" ADD COLUMN expired_at DATETIME"); err != nil {
g.Log().Warningf(ctx, "add column expired_at failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "ALTER TABLE "+consts.TableNameAgentProfile+" ADD COLUMN region_protected INTEGER NOT NULL DEFAULT 0"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "ALTER TABLE "+consts.TableNameAgentProfile+" ADD COLUMN region_protected INTEGER NOT NULL DEFAULT 0"); err != nil {
g.Log().Warningf(ctx, "add column region_protected failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "ALTER TABLE "+consts.TableNameAgentProfile+" DROP COLUMN tier_id"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "ALTER TABLE "+consts.TableNameAgentProfile+" DROP COLUMN tier_id"); err != nil {
g.Log().Warningf(ctx, "drop column tier_id failed: %v", err)
}
@@ -104,7 +161,7 @@ func clearAgentProfileCache(ctx context.Context, userId int64) {
func (d *agentProfileDao) Get(ctx context.Context, userId int64) (*entity.AgentProfile, error) {
var p entity.AgentProfile
err := g.DB().Model(consts.TableNameAgentProfile).Ctx(ctx).
err := g.DB(consts.DbGroupFinance).Model(consts.TableNameAgentProfile).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: "agentProfile_Get_" + gconv.String(userId)}).
Where("user_id", userId).Scan(&p)
if err != nil {
@@ -121,7 +178,7 @@ func (d *agentProfileDao) Save(ctx context.Context, data *entity.AgentProfile) e
if data.ExpiredAt != nil {
expiredAtStr = data.ExpiredAt.Format("Y-m-d H:i:s")
}
_, err := g.DB().Exec(ctx,
_, err := g.DB(consts.DbGroupFinance).Exec(ctx,
"INSERT OR REPLACE INTO "+consts.TableNameAgentProfile+" (user_id, max_customers, renewals, expired_at, region_protected) VALUES (?, ?, ?, ?, ?)",
data.UserId, data.MaxCustomers, data.Renewals, expiredAtStr, data.RegionProtected)
clearAgentProfileCache(ctx, data.UserId)
@@ -137,7 +194,7 @@ func (d *agentProfileDao) Update(ctx context.Context, data *entity.AgentProfile)
if data.ExpiredAt != nil {
m["expired_at"] = data.ExpiredAt.Format("Y-m-d H:i:s")
}
_, err := g.DB().Model(consts.TableNameAgentProfile).Ctx(ctx).Data(m).Where("user_id", data.UserId).Update()
_, err := g.DB(consts.DbGroupFinance).Model(consts.TableNameAgentProfile).Ctx(ctx).Data(m).Where("user_id", data.UserId).Update()
clearAgentProfileCache(ctx, data.UserId)
return err
}
shortdrama/dao/customer_profile_dao.go
+157 -40
View File
@@ -32,55 +32,172 @@ type customerListRow struct {
}
// ListCustomerWithAgent 按条件分页查询客户列表（含代理商名）
// 拆库后改为多个单表查询 + 内存合并，避免跨表 JOIN
func (d *customerProfileDao) ListCustomerWithAgent(ctx context.Context, agentId int64, keyword, phone, province, region, agentName string, page, pageSize int) ([]*customerListRow, int, error) {
m := g.DB().Model(consts.TableNameUser+" u").
InnerJoin(consts.TableNameCustomerProfile+" cp", "cp.user_id = u.id").
LeftJoin(consts.TableNameUser+" au", "au.id = cp.agent_id").
Where("u.role", "customer")
// 分页+筛选查询不缓存（参数不同导致缓存键碰撞）
if agentId > 0 {
m = m.Where("cp.agent_id", agentId)
}
if keyword != "" {
m = m.Where("u.name LIKE ? OR u.phone LIKE ?", "%"+keyword+"%", "%"+keyword+"%")
}
if phone != "" {
m = m.Where("u.phone LIKE ?", "%"+phone+"%")
}
if province != "" {
m = m.Where("u.province = ?", province)
}
if region != "" {
m = m.Where("u.region = ?", region)
}
if agentName != "" {
m = m.Where("au.name LIKE ? OR au.username LIKE ? OR au.phone LIKE ?", "%"+agentName+"%", "%"+agentName+"%", "%"+agentName+"%")
}
if page < 1 {
page = 1
}
if pageSize < 1 {
pageSize = 20
}
m = m.Fields(
"u.id", "u.phone", "u.name", "u.province", "u.region", "u.address", "u.created_at",
"cp.agent_id", "cp.balance",
"au.name agent_name",
)
r, total, err := m.OrderAsc("u.id").Limit(pageSize).Offset((page - 1) * pageSize).AllAndCount(false)
// 代理商维度筛选：单表查出符合条件的客户 user_id 集合
var customerIds []int64
switch {
case agentId > 0:
ids, err := d.listUserIdsByAgent(ctx, agentId)
if err != nil {
return nil, 0, err
}
if len(ids) == 0 {
return make([]*customerListRow, 0), 0, nil
}
customerIds = ids
case agentName != "":
agentIds, err := d.listAgentIdsByName(ctx, agentName)
if err != nil {
return nil, 0, err
}
if len(agentIds) == 0 {
return make([]*customerListRow, 0), 0, nil
}
ids, err := d.listUserIdsByAgents(ctx, agentIds)
if err != nil {
return nil, 0, err
}
if len(ids) == 0 {
return make([]*customerListRow, 0), 0, nil
}
customerIds = ids
}
// 客户主查询（user 表，role=customer + 关键字/手机号/省市筛选）
m := g.DB(consts.DbGroupFinance).Model(consts.TableNameUser).Ctx(ctx).Where("role", "customer")
if keyword != "" {
m = m.Where("name LIKE ? OR phone LIKE ?", "%"+keyword+"%", "%"+keyword+"%")
}
if phone != "" {
m = m.Where("phone LIKE ?", "%"+phone+"%")
}
if province != "" {
m = m.Where("province = ?", province)
}
if region != "" {
m = m.Where("region = ?", region)
}
if len(customerIds) > 0 {
m = m.WhereIn("id", customerIds)
}
total, err := m.Count()
if err != nil {
return nil, 0, err
}
var rows []*customerListRow
err = r.Structs(&rows)
return rows, total, err
var users []*entity.User
err = m.OrderAsc("id").Limit(pageSize).Offset((page - 1) * pageSize).Scan(&users)
if err != nil {
return nil, 0, err
}
if len(users) == 0 {
return make([]*customerListRow, 0), total, nil
}
// 批量补客户档案（agent_id、balance）
pageIds := make([]int64, 0, len(users))
for _, u := range users {
pageIds = append(pageIds, u.Id)
}
profiles := make(map[int64]*entity.CustomerProfile)
var profileList []*entity.CustomerProfile
if err := g.DB(consts.DbGroupFinance).Model(consts.TableNameCustomerProfile).Ctx(ctx).
WhereIn("user_id", pageIds).Scan(&profileList); err != nil {
return nil, 0, err
}
for _, p := range profileList {
profiles[p.UserId] = p
}
// 批量补代理商名称
agentNameMap := make(map[int64]string)
agentIds := make([]int64, 0, len(profiles))
for _, p := range profiles {
if p.AgentId > 0 {
agentIds = append(agentIds, p.AgentId)
}
}
if len(agentIds) > 0 {
var agents []*entity.User
if err := g.DB(consts.DbGroupFinance).Model(consts.TableNameUser).Ctx(ctx).
Fields("id", "name").WhereIn("id", agentIds).Scan(&agents); err != nil {
return nil, 0, err
}
for _, a := range agents {
agentNameMap[a.Id] = a.Name
}
}
// 内存合并组装结果
rows := make([]*customerListRow, 0, len(users))
for _, u := range users {
row := &customerListRow{
Id: u.Id,
Phone: u.Phone,
Name: u.Name,
Province: u.Province,
Region: u.Region,
Address: u.Address,
CreatedAt: u.CreatedAt,
}
if p, ok := profiles[u.Id]; ok {
row.AgentId = p.AgentId
row.Balance = p.Balance
row.AgentName = agentNameMap[p.AgentId]
}
rows = append(rows, row)
}
return rows, total, nil
}
// listAgentIdsByName 按名称/账号/手机号模糊匹配代理商 user_id
func (d *customerProfileDao) listAgentIdsByName(ctx context.Context, keyword string) ([]int64, error) {
r, err := g.DB(consts.DbGroupFinance).Model(consts.TableNameUser).Ctx(ctx).
Fields("id").Where("role", "agent").
Where("name LIKE ? OR username LIKE ? OR phone LIKE ?", "%"+keyword+"%", "%"+keyword+"%", "%"+keyword+"%").
All()
if err != nil {
return nil, err
}
ids := make([]int64, 0, len(r))
for _, row := range r {
ids = append(ids, row["id"].Int64())
}
return ids, nil
}
// listUserIdsByAgent 查询指定代理商名下的客户 user_id
func (d *customerProfileDao) listUserIdsByAgent(ctx context.Context, agentId int64) ([]int64, error) {
return d.listUserIdsByAgents(ctx, []int64{agentId})
}
// listUserIdsByAgents 批量查询多个代理商名下的客户 user_id
func (d *customerProfileDao) listUserIdsByAgents(ctx context.Context, agentIds []int64) ([]int64, error) {
if len(agentIds) == 0 {
return nil, nil
}
r, err := g.DB(consts.DbGroupFinance).Model(consts.TableNameCustomerProfile).Ctx(ctx).
Fields("user_id").WhereIn("agent_id", agentIds).All()
if err != nil {
return nil, err
}
ids := make([]int64, 0, len(r))
for _, row := range r {
ids = append(ids, row["user_id"].Int64())
}
return ids, nil
}
func init() {
ctx := context.Background()
_, err := g.DB().Exec(ctx, `CREATE TABLE IF NOT EXISTS `+consts.TableNameCustomerProfile+` (
_, err := g.DB(consts.DbGroupFinance).Exec(ctx, `CREATE TABLE IF NOT EXISTS `+consts.TableNameCustomerProfile+` (
user_id INTEGER PRIMARY KEY,
agent_id INTEGER NOT NULL DEFAULT 0,
balance INTEGER NOT NULL DEFAULT 0
@@ -88,7 +205,7 @@ func init() {
if err != nil {
g.Log().Warningf(ctx, "create customer_profile table failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "CREATE INDEX IF NOT EXISTS idx_cp_agent ON "+consts.TableNameCustomerProfile+"(agent_id)"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "CREATE INDEX IF NOT EXISTS idx_cp_agent ON "+consts.TableNameCustomerProfile+"(agent_id)"); err != nil {
g.Log().Warningf(ctx, "create index idx_cp_agent failed: %v", err)
}
}
@@ -101,7 +218,7 @@ func clearCustomerProfileCache(ctx context.Context, userId int64, agentId int64)
func (d *customerProfileDao) Get(ctx context.Context, userId int64) (*entity.CustomerProfile, error) {
var p entity.CustomerProfile
err := g.DB().Model(consts.TableNameCustomerProfile).Ctx(ctx).
err := g.DB(consts.DbGroupFinance).Model(consts.TableNameCustomerProfile).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: "customerProfile_Get_" + gconv.String(userId)}).
Where("user_id", userId).Scan(&p)
if err != nil {
@@ -114,7 +231,7 @@ func (d *customerProfileDao) Get(ctx context.Context, userId int64) (*entity.Cus
}
func (d *customerProfileDao) Save(ctx context.Context, data *entity.CustomerProfile) error {
_, err := g.DB().Exec(ctx,
_, err := g.DB(consts.DbGroupFinance).Exec(ctx,
"INSERT OR REPLACE INTO "+consts.TableNameCustomerProfile+" (user_id, agent_id, balance) VALUES (?, ?, ?)",
data.UserId, data.AgentId, data.Balance)
clearCustomerProfileCache(ctx, data.UserId, data.AgentId)
@@ -127,14 +244,14 @@ func (d *customerProfileDao) UpdateBalance(ctx context.Context, userId int64, ne
if p, _ := d.Get(ctx, userId); p != nil {
agentId = p.AgentId
}
_, err := g.DB().Model(consts.TableNameCustomerProfile).Ctx(ctx).
_, err := g.DB(consts.DbGroupFinance).Model(consts.TableNameCustomerProfile).Ctx(ctx).
Data(g.Map{"balance": newBalance}).Where("user_id", userId).Update()
clearCustomerProfileCache(ctx, userId, agentId)
return err
}
func (d *customerProfileDao) CountByAgent(ctx context.Context, agentId int64) (int, error) {
c, err := g.DB().Model(consts.TableNameCustomerProfile).Ctx(ctx).
c, err := g.DB(consts.DbGroupFinance).Model(consts.TableNameCustomerProfile).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: "customerProfile_CountByAgent_" + gconv.String(agentId)}).
Where("agent_id", agentId).Count()
return c, err
shortdrama/dao/model_config_dao.go
+45 -28
View File
@@ -20,7 +20,7 @@ type modelConfigDao struct{}
func init() {
ctx := context.Background()
_, err := g.DB().Exec(ctx,
_, err := g.DB(consts.DbGroupSystem).Exec(ctx,
"CREATE TABLE IF NOT EXISTS "+consts.TableNameModelConfig+" ("+
"id INTEGER PRIMARY KEY AUTOINCREMENT,"+
"model_type TEXT NOT NULL DEFAULT '',"+
@@ -38,7 +38,9 @@ func init() {
// 检测旧表结构：含 base_url / api_key / chat_api_key / is_active / task_callback_url 等旧列 → 迁移
hasOldColumns := false
if r, _ := g.DB().GetAll(ctx, "PRAGMA table_info("+consts.TableNameModelConfig+")"); r != nil {
if r, qErr := g.DB(consts.DbGroupSystem).GetAll(ctx, "PRAGMA table_info("+consts.TableNameModelConfig+")"); qErr != nil {
g.Log().Warningf(ctx, "query model_config table_info failed: %v", qErr)
} else if r != nil {
for _, col := range r {
name := col["name"].String()
if name == "base_url" || name == "api_key" || name == "chat_api_key" || name == "is_active" || name == "price_per_second" || name == "max_tokens" || name == "task_callback_url" {
@@ -51,7 +53,7 @@ func init() {
if hasOldColumns {
g.Log().Info(ctx, "检测到旧表结构，开始迁移 model_config...")
newTable := consts.TableNameModelConfig + "_new"
if _, err := g.DB().Exec(ctx,
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx,
"CREATE TABLE IF NOT EXISTS "+newTable+" ("+
"id INTEGER PRIMARY KEY AUTOINCREMENT,"+
"model_type TEXT NOT NULL DEFAULT '',"+
@@ -68,15 +70,18 @@ func init() {
}
hasChatApiKey := false
if _, err := g.DB().Exec(ctx, "SELECT chat_api_key FROM "+consts.TableNameModelConfig+" LIMIT 1"); err == nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, "SELECT chat_api_key FROM "+consts.TableNameModelConfig+" LIMIT 1"); err == nil {
hasChatApiKey = true
}
now := gtime.Now().Format("Y-m-d H:i:s")
if hasChatApiKey {
oldRow, _ := g.DB().Model(consts.TableNameModelConfig).Ctx(ctx).Limit(1).One()
oldRow, qErr := g.DB(consts.DbGroupSystem).Model(consts.TableNameModelConfig).Ctx(ctx).Limit(1).One()
if qErr != nil {
g.Log().Warningf(ctx, "query old model config row failed: %v", qErr)
}
if oldRow != nil && !oldRow.IsEmpty() {
if _, err := g.DB().Model(newTable).Ctx(ctx).Data(g.Map{
if _, err := g.DB(consts.DbGroupSystem).Model(newTable).Ctx(ctx).Data(g.Map{
"model_type": "chat", "model_name": oldRow["chat_model_name"],
"schema": oldRow["chat_schema"],
"price": 0, "price_unit": "video",
@@ -84,7 +89,7 @@ func init() {
}).Insert(); err != nil {
g.Log().Warningf(ctx, "insert chat model config failed: %v", err)
}
if _, err := g.DB().Model(newTable).Ctx(ctx).Data(g.Map{
if _, err := g.DB(consts.DbGroupSystem).Model(newTable).Ctx(ctx).Data(g.Map{
"model_type": "video", "model_name": oldRow["video_model_name"],
"schema": oldRow["video_schema"],
"price": gconv.Int(oldRow["price_per_second"]), "price_unit": "second",
@@ -94,7 +99,10 @@ func init() {
}
}
} else {
oldRows, _ := g.DB().Model(consts.TableNameModelConfig).Ctx(ctx).OrderAsc("id").All()
oldRows, qErr := g.DB(consts.DbGroupSystem).Model(consts.TableNameModelConfig).Ctx(ctx).OrderAsc("id").All()
if qErr != nil {
g.Log().Warningf(ctx, "query old model config rows failed: %v", qErr)
}
for _, or := range oldRows {
mt := or["model_type"].String()
priceVal := gconv.Int(or["price_per_second"])
@@ -105,7 +113,7 @@ func init() {
if mt == "chat" {
unit = "video"
}
if _, err := g.DB().Model(newTable).Ctx(ctx).Data(g.Map{
if _, err := g.DB(consts.DbGroupSystem).Model(newTable).Ctx(ctx).Data(g.Map{
"model_type": mt, "model_name": or["model_name"],
"schema": or["schema"],
"price": priceVal, "price_unit": unit,
@@ -115,17 +123,19 @@ func init() {
}
}
}
if _, err := g.DB().Exec(ctx, "DROP TABLE "+consts.TableNameModelConfig); err != nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, "DROP TABLE "+consts.TableNameModelConfig); err != nil {
g.Log().Warningf(ctx, "drop old model_config table failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "ALTER TABLE "+newTable+" RENAME TO "+consts.TableNameModelConfig); err != nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, "ALTER TABLE "+newTable+" RENAME TO "+consts.TableNameModelConfig); err != nil {
g.Log().Warningf(ctx, "rename temp table to model_config failed: %v", err)
}
g.Log().Info(ctx, "model_config 表结构迁移完成")
}
// 检测缺少 concurrency_count 列的情况（已有新表结构但字段不全）
if r, _ := g.DB().GetAll(ctx, "PRAGMA table_info("+consts.TableNameModelConfig+")"); r != nil {
if r, qErr := g.DB(consts.DbGroupSystem).GetAll(ctx, "PRAGMA table_info("+consts.TableNameModelConfig+")"); qErr != nil {
g.Log().Warningf(ctx, "query model_config table_info failed: %v", qErr)
} else if r != nil {
hasCol := false
for _, col := range r {
if col["name"].String() == "concurrency_count" {
@@ -135,7 +145,7 @@ func init() {
}
if !hasCol {
g.Log().Info(ctx, "检测到 model_config 缺少 concurrency_count 列，正在补充...")
if _, err := g.DB().Exec(ctx, "ALTER TABLE "+consts.TableNameModelConfig+" ADD COLUMN concurrency_count INTEGER NOT NULL DEFAULT 1"); err != nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, "ALTER TABLE "+consts.TableNameModelConfig+" ADD COLUMN concurrency_count INTEGER NOT NULL DEFAULT 1"); err != nil {
g.Log().Warningf(ctx, "add concurrency_count column failed: %v", err)
} else {
g.Log().Info(ctx, "concurrency_count 列补充完成")
@@ -144,7 +154,9 @@ func init() {
}
// 检测并迁移：first_frame_mapping -> schema_mapping
if r, _ := g.DB().GetAll(ctx, "PRAGMA table_info("+consts.TableNameModelConfig+")"); r != nil {
if r, qErr := g.DB(consts.DbGroupSystem).GetAll(ctx, "PRAGMA table_info("+consts.TableNameModelConfig+")"); qErr != nil {
g.Log().Warningf(ctx, "query model_config table_info failed: %v", qErr)
} else if r != nil {
hasFirstFrame := false
hasSchemaMapping := false
for _, col := range r {
@@ -159,18 +171,18 @@ func init() {
// 旧表重命名 first_frame_mapping -> schema_mapping
if hasFirstFrame && !hasSchemaMapping {
g.Log().Info(ctx, "检测到旧 first_frame_mapping 列，正在迁移到 schema_mapping...")
if _, err := g.DB().Exec(ctx, "ALTER TABLE "+consts.TableNameModelConfig+" RENAME COLUMN first_frame_mapping TO schema_mapping"); err != nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, "ALTER TABLE "+consts.TableNameModelConfig+" RENAME COLUMN first_frame_mapping TO schema_mapping"); err != nil {
g.Log().Warningf(ctx, "rename first_frame_mapping to schema_mapping failed: %v", err)
} else {
// 将旧值（纯文本路径）转为 JSON 格式
g.DB().Exec(ctx, "UPDATE "+consts.TableNameModelConfig+" SET schema_mapping = '{}' WHERE schema_mapping != '' AND schema_mapping NOT LIKE '{'")
g.DB(consts.DbGroupSystem).Exec(ctx, "UPDATE "+consts.TableNameModelConfig+" SET schema_mapping = '{}' WHERE schema_mapping != '' AND schema_mapping NOT LIKE '{'")
g.Log().Info(ctx, "first_frame_mapping 迁移到 schema_mapping 完成")
}
}
// 无任何映射列时直接添加 schema_mapping
if !hasFirstFrame && !hasSchemaMapping {
g.Log().Info(ctx, "检测到 model_config 缺少 schema_mapping 列，正在补充...")
if _, err := g.DB().Exec(ctx, "ALTER TABLE "+consts.TableNameModelConfig+" ADD COLUMN schema_mapping TEXT NOT NULL DEFAULT ''"); err != nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, "ALTER TABLE "+consts.TableNameModelConfig+" ADD COLUMN schema_mapping TEXT NOT NULL DEFAULT ''"); err != nil {
g.Log().Warningf(ctx, "add schema_mapping column failed: %v", err)
} else {
g.Log().Info(ctx, "schema_mapping 列补充完成")
@@ -179,7 +191,9 @@ func init() {
}
// 检测并删除已废弃的 reference_template 列（后续从 schema_mapping 中获取）
if r, _ := g.DB().GetAll(ctx, "PRAGMA table_info("+consts.TableNameModelConfig+")"); r != nil {
if r, qErr := g.DB(consts.DbGroupSystem).GetAll(ctx, "PRAGMA table_info("+consts.TableNameModelConfig+")"); qErr != nil {
g.Log().Warningf(ctx, "query model_config table_info failed: %v", qErr)
} else if r != nil {
hasRefTemplate := false
for _, col := range r {
if col["name"].String() == "reference_template" {
@@ -189,7 +203,7 @@ func init() {
}
if hasRefTemplate {
g.Log().Info(ctx, "检测到已废弃的 reference_template 列，正在删除...")
if _, err := g.DB().Exec(ctx, "ALTER TABLE "+consts.TableNameModelConfig+" DROP COLUMN reference_template"); err != nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, "ALTER TABLE "+consts.TableNameModelConfig+" DROP COLUMN reference_template"); err != nil {
g.Log().Warningf(ctx, "drop reference_template column failed: %v", err)
} else {
g.Log().Info(ctx, "reference_template 列删除完成")
@@ -198,7 +212,10 @@ func init() {
}
// 初始化默认模型配置（仅当表为空时）
count, _ := g.DB().Model(consts.TableNameModelConfig).Ctx(ctx).Count()
count, qErr := g.DB(consts.DbGroupSystem).Model(consts.TableNameModelConfig).Ctx(ctx).Count()
if qErr != nil {
g.Log().Warningf(ctx, "count model config failed: %v", qErr)
}
if count == 0 {
now := gtime.Now().Format("Y-m-d H:i:s")
defaults := []g.Map{
@@ -206,7 +223,7 @@ func init() {
{"model_type": "video", "model_name": "", "schema": "", "price": 10, "price_unit": "second", "created_at": now, "updated_at": now},
}
for _, m := range defaults {
if _, e := g.DB().Model(consts.TableNameModelConfig).Ctx(ctx).Data(m).Insert(); e != nil {
if _, e := g.DB(consts.DbGroupSystem).Model(consts.TableNameModelConfig).Ctx(ctx).Data(m).Insert(); e != nil {
g.Log().Warningf(ctx, "init default model config failed: %v", e)
}
}
@@ -224,7 +241,7 @@ func clearModelConfigCache(ctx context.Context, modelType string) {
}
func (d *modelConfigDao) ListPage(ctx context.Context, page, pageSize int, modelType, keyword string) (res []*entity.ModelConfig, total int, err error) {
m := g.DB().Model(consts.TableNameModelConfig).Ctx(ctx)
m := g.DB(consts.DbGroupSystem).Model(consts.TableNameModelConfig).Ctx(ctx)
if modelType != "" {
m = m.Where("model_type", modelType)
}
@@ -270,7 +287,7 @@ func (d *modelConfigDao) ListPage(ctx context.Context, page, pageSize int, model
}
func (d *modelConfigDao) GetFirst(ctx context.Context) (res *entity.ModelConfig, err error) {
r, err := g.DB().Model(consts.TableNameModelConfig).Ctx(ctx).
r, err := g.DB(consts.DbGroupSystem).Model(consts.TableNameModelConfig).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: "modelConfig_GetFirst"}).
OrderAsc("id").Limit(1).One()
if err != nil {
@@ -285,7 +302,7 @@ func (d *modelConfigDao) GetFirst(ctx context.Context) (res *entity.ModelConfig,
}
func (d *modelConfigDao) GetActiveModelByType(ctx context.Context, modelType string) (res *entity.ModelConfig, err error) {
r, err := g.DB().Model(consts.TableNameModelConfig).Ctx(ctx).
r, err := g.DB(consts.DbGroupSystem).Model(consts.TableNameModelConfig).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: "modelConfig_GetActiveModelByType_" + modelType}).
Where("model_type", modelType).Limit(1).One()
if err != nil {
@@ -303,7 +320,7 @@ func (d *modelConfigDao) GetByIds(ctx context.Context, ids []int64) (res []*enti
if len(ids) == 0 {
return make([]*entity.ModelConfig, 0), nil
}
r, err := g.DB().Model(consts.TableNameModelConfig).Ctx(ctx).
r, err := g.DB(consts.DbGroupSystem).Model(consts.TableNameModelConfig).Ctx(ctx).
WhereIn("id", ids).OrderAsc("id").All()
if err != nil {
return nil, err
@@ -316,7 +333,7 @@ func (d *modelConfigDao) GetByIds(ctx context.Context, ids []int64) (res []*enti
}
func (d *modelConfigDao) GetAll(ctx context.Context) (res []*entity.ModelConfig, err error) {
r, err := g.DB().Model(consts.TableNameModelConfig).Ctx(ctx).
r, err := g.DB(consts.DbGroupSystem).Model(consts.TableNameModelConfig).Ctx(ctx).
Cache(gdb.CacheOption{Duration: time.Minute, Name: "model_config_GetAll"}).
OrderAsc("id").Limit(200).All()
if err != nil {
@@ -331,7 +348,7 @@ func (d *modelConfigDao) GetAll(ctx context.Context) (res []*entity.ModelConfig,
func (d *modelConfigDao) Save(ctx context.Context, data *entity.ModelConfig) error {
if data.Id > 0 {
_, err := g.DB().Model(consts.TableNameModelConfig).Ctx(ctx).
_, err := g.DB(consts.DbGroupSystem).Model(consts.TableNameModelConfig).Ctx(ctx).
Data(data).Where("id", data.Id).Update()
if err != nil {
return err
@@ -341,7 +358,7 @@ func (d *modelConfigDao) Save(ctx context.Context, data *entity.ModelConfig) err
delete(m, "id")
m["created_at"] = gtime.Now().Format("Y-m-d H:i:s")
m["updated_at"] = gtime.Now().Format("Y-m-d H:i:s")
_, err := g.DB().Model(consts.TableNameModelConfig).Ctx(ctx).
_, err := g.DB(consts.DbGroupSystem).Model(consts.TableNameModelConfig).Ctx(ctx).
Data(m).Insert()
if err != nil {
return err
shortdrama/dao/payment_channel_trade_dao.go
+4 -4
View File
@@ -18,7 +18,7 @@ type paymentChannelTradeDao struct{}
func init() {
ctx := context.Background()
_, err := g.DB().Exec(ctx, `CREATE TABLE IF NOT EXISTS `+consts.TableNamePaymentChannelTrade+` (
_, err := g.DB(consts.DbGroupFinance).Exec(ctx, `CREATE TABLE IF NOT EXISTS `+consts.TableNamePaymentChannelTrade+` (
id INTEGER PRIMARY KEY AUTOINCREMENT,
order_id INTEGER NOT NULL DEFAULT 0,
channel TEXT NOT NULL DEFAULT '',
@@ -31,7 +31,7 @@ func init() {
if err != nil {
g.Log().Warningf(ctx, "create payment_channel_trade table failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "CREATE INDEX IF NOT EXISTS idx_pct_order ON "+consts.TableNamePaymentChannelTrade+"(order_id)"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "CREATE INDEX IF NOT EXISTS idx_pct_order ON "+consts.TableNamePaymentChannelTrade+"(order_id)"); err != nil {
g.Log().Warningf(ctx, "create index idx_pct_order failed: %v", err)
}
}
@@ -41,7 +41,7 @@ func clearPaymentChannelTradeCache(ctx context.Context, orderId int64) {
}
func (d *paymentChannelTradeDao) Insert(ctx context.Context, data *entity.PaymentChannelTrade) (int64, error) {
r, err := g.DB().Exec(ctx,
r, err := g.DB(consts.DbGroupFinance).Exec(ctx,
"INSERT INTO "+consts.TableNamePaymentChannelTrade+" (order_id, channel, prepay_id, code_url, trade_no, channel_response, created_at) VALUES (?, ?, ?, ?, ?, ?, datetime('now','localtime'))",
data.OrderId, data.Channel, data.PrepayId, data.CodeUrl, data.TradeNo, data.ChannelResponse)
if err != nil {
@@ -53,7 +53,7 @@ func (d *paymentChannelTradeDao) Insert(ctx context.Context, data *entity.Paymen
func (d *paymentChannelTradeDao) GetByOrderId(ctx context.Context, orderId int64) (*entity.PaymentChannelTrade, error) {
var t entity.PaymentChannelTrade
err := g.DB().Model(consts.TableNamePaymentChannelTrade).Ctx(ctx).
err := g.DB(consts.DbGroupFinance).Model(consts.TableNamePaymentChannelTrade).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: "paymentChannelTrade_GetByOrderId_" + gconv.String(orderId)}).
Where("order_id", orderId).Scan(&t)
if err != nil {
shortdrama/dao/payment_config_dao.go
+27 -6
View File
@@ -20,7 +20,7 @@ type paymentConfigDao struct{}
func init() {
ctx := context.Background()
if _, err := g.DB().Exec(ctx, `CREATE TABLE IF NOT EXISTS `+consts.TableNamePaymentConfig+` (
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, `CREATE TABLE IF NOT EXISTS `+consts.TableNamePaymentConfig+` (
id INTEGER PRIMARY KEY AUTOINCREMENT,
channel TEXT NOT NULL DEFAULT '',
channel_type TEXT NOT NULL DEFAULT '',
@@ -35,6 +35,25 @@ func init() {
)`); err != nil {
g.Log().Warningf(ctx, "create payment config table failed: %v", err)
}
// backfill created_at for legacy data
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, "UPDATE "+consts.TableNamePaymentConfig+" SET created_at = datetime('now','localtime') WHERE created_at IS NULL"); err != nil {
g.Log().Warningf(ctx, "backfill payment_config created_at failed: %v", err)
}
// UTC to local timezone migration, run once via PRAGMA user_version
var userVersion int
if r, qErr := g.DB(consts.DbGroupSystem).Query(ctx, "PRAGMA user_version"); qErr != nil {
g.Log().Warningf(ctx, "query user_version failed: %v", qErr)
} else if len(r) > 0 {
userVersion = r[0]["user_version"].Int()
}
if userVersion == 0 {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, "UPDATE "+consts.TableNamePaymentConfig+" SET created_at = datetime(created_at, '+8 hours') WHERE created_at LIKE '%-%T%:%.%Z' OR created_at LIKE '%-%T%:%s%z'"); err != nil {
g.Log().Warningf(ctx, "migrate payment_config UTC to local failed: %v", err)
}
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, "PRAGMA user_version = 1"); err != nil {
g.Log().Warningf(ctx, "set user_version failed: %v", err)
}
}
}
func clearPaymentConfigCache(ctx context.Context) {
@@ -47,7 +66,7 @@ func (d *paymentConfigDao) GetByChannel(ctx context.Context, channel string, typ
if len(typeOpt) > 0 && typeOpt[0] != "" {
cacheKey += "_" + typeOpt[0]
}
m := g.DB().Model(consts.TableNamePaymentConfig).Ctx(ctx).
m := g.DB(consts.DbGroupSystem).Model(consts.TableNamePaymentConfig).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: cacheKey}).
Where("channel", channel)
if len(typeOpt) > 0 && typeOpt[0] != "" {
@@ -72,7 +91,7 @@ func (d *paymentConfigDao) Save(ctx context.Context, data *entity.PaymentConfig)
}
if existing != nil {
data.Id = existing.Id
_, err = g.DB().Model(consts.TableNamePaymentConfig).Ctx(ctx).
_, err = g.DB(consts.DbGroupSystem).Model(consts.TableNamePaymentConfig).Ctx(ctx).
Data(data).Where("id", existing.Id).Update()
clearPaymentConfigCache(ctx)
return err
@@ -81,14 +100,14 @@ func (d *paymentConfigDao) Save(ctx context.Context, data *entity.PaymentConfig)
delete(m, "id")
m["created_at"] = gtime.Now().Format("Y-m-d H:i:s")
m["updated_at"] = gtime.Now().Format("Y-m-d H:i:s")
_, err = g.DB().Model(consts.TableNamePaymentConfig).Ctx(ctx).
_, err = g.DB(consts.DbGroupSystem).Model(consts.TableNamePaymentConfig).Ctx(ctx).
Data(m).Insert()
clearPaymentConfigCache(ctx)
return err
}
func (d *paymentConfigDao) GetAll(ctx context.Context) (res []*entity.PaymentConfig, err error) {
r, err := g.DB().Model(consts.TableNamePaymentConfig).Ctx(ctx).
r, err := g.DB(consts.DbGroupSystem).Model(consts.TableNamePaymentConfig).Ctx(ctx).
Cache(gdb.CacheOption{Duration: time.Minute, Name: "payment_config_GetAll"}).
OrderAsc("id").All()
if err != nil {
@@ -99,7 +118,9 @@ func (d *paymentConfigDao) GetAll(ctx context.Context) (res []*entity.PaymentCon
}
for _, item := range r {
cfg := new(entity.PaymentConfig)
_ = item.Struct(&cfg)
if err := item.Struct(&cfg); err != nil {
g.Log().Warningf(ctx, "parse payment config row failed: %v", err)
}
res = append(res, cfg)
}
return
shortdrama/dao/payment_order_dao.go
+12 -12
View File
@@ -18,7 +18,7 @@ type paymentOrderDao struct{}
func init() {
ctx := context.Background()
_, err := g.DB().Exec(ctx, `CREATE TABLE IF NOT EXISTS `+consts.TableNamePaymentOrder+` (
_, err := g.DB(consts.DbGroupFinance).Exec(ctx, `CREATE TABLE IF NOT EXISTS `+consts.TableNamePaymentOrder+` (
id INTEGER PRIMARY KEY AUTOINCREMENT,
order_no TEXT NOT NULL UNIQUE,
order_type TEXT NOT NULL DEFAULT 'recharge',
@@ -37,19 +37,19 @@ func init() {
if err != nil {
g.Log().Warningf(ctx, "create payment_order table failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "CREATE INDEX IF NOT EXISTS idx_po_user ON "+consts.TableNamePaymentOrder+"(user_id)"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "CREATE INDEX IF NOT EXISTS idx_po_user ON "+consts.TableNamePaymentOrder+"(user_id)"); err != nil {
g.Log().Warningf(ctx, "create index idx_po_user failed: %v", err)
}
if _, err := g.DB().Exec(ctx,
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx,
"CREATE INDEX IF NOT EXISTS idx_po_order_type ON payment_order(order_type)"); err != nil {
g.Log().Warningf(ctx, "create order type index failed: %v", err)
}
if _, err := g.DB().Exec(ctx,
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx,
"CREATE INDEX IF NOT EXISTS idx_po_status ON payment_order(status)"); err != nil {
g.Log().Warningf(ctx, "create order status index failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "ALTER TABLE "+consts.TableNamePaymentOrder+" ADD COLUMN order_type TEXT NOT NULL DEFAULT 'recharge'"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "ALTER TABLE "+consts.TableNamePaymentOrder+" ADD COLUMN order_type TEXT NOT NULL DEFAULT 'recharge'"); err != nil {
g.Log().Warningf(ctx, "add column order_type failed: %v", err)
}
}
@@ -62,7 +62,7 @@ func clearPaymentOrderCache(ctx context.Context, userId int64, status string) {
}
func (d *paymentOrderDao) Insert(ctx context.Context, data *entity.PaymentOrder) (int64, error) {
r, err := g.DB().Exec(ctx,
r, err := g.DB(consts.DbGroupFinance).Exec(ctx,
"INSERT INTO "+consts.TableNamePaymentOrder+" (order_no, order_type, user_id, amount, channel, channel_type, status, subject, notify_raw, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now','localtime'), datetime('now','localtime'))",
data.OrderNo, data.OrderType, data.UserId, data.Amount, data.Channel, data.ChannelType, data.Status, data.Subject, data.NotifyRaw)
if err != nil {
@@ -74,7 +74,7 @@ func (d *paymentOrderDao) Insert(ctx context.Context, data *entity.PaymentOrder)
func (d *paymentOrderDao) GetByOrderNo(ctx context.Context, orderNo string) (*entity.PaymentOrder, error) {
var p entity.PaymentOrder
err := g.DB().Model(consts.TableNamePaymentOrder).Ctx(ctx).
err := g.DB(consts.DbGroupFinance).Model(consts.TableNamePaymentOrder).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: "paymentOrder_GetByOrderNo_" + orderNo}).
Where("order_no", orderNo).Scan(&p)
if err != nil {
@@ -88,14 +88,14 @@ func (d *paymentOrderDao) GetByOrderNo(ctx context.Context, orderNo string) (*en
func (d *paymentOrderDao) ListByStatus(ctx context.Context, status string) ([]*entity.PaymentOrder, error) {
var list []*entity.PaymentOrder
err := g.DB().Model(consts.TableNamePaymentOrder).Ctx(ctx).
err := g.DB(consts.DbGroupFinance).Model(consts.TableNamePaymentOrder).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: "paymentOrder_ListByStatus_" + status}).
Where("status", status).OrderDesc("id").Limit(200).Scan(&list)
return list, err
}
func (d *paymentOrderDao) UpdateSuccess(ctx context.Context, id int64, notifyRaw string) error {
_, err := g.DB().Exec(ctx,
_, err := g.DB(consts.DbGroupFinance).Exec(ctx,
"UPDATE "+consts.TableNamePaymentOrder+" SET status='success', paid_at=datetime('now','localtime'), notify_raw=?, notify_count=notify_count+1, updated_at=datetime('now','localtime') WHERE id=?",
notifyRaw, id)
// Can only clear list-by-status cache
@@ -105,7 +105,7 @@ func (d *paymentOrderDao) UpdateSuccess(ctx context.Context, id int64, notifyRaw
}
func (d *paymentOrderDao) UpdateFail(ctx context.Context, id int64, notifyRaw string) error {
_, err := g.DB().Exec(ctx,
_, err := g.DB(consts.DbGroupFinance).Exec(ctx,
"UPDATE "+consts.TableNamePaymentOrder+" SET status='failed', notify_raw=?, notify_count=notify_count+1, updated_at=datetime('now','localtime') WHERE id=?",
notifyRaw, id)
_, _ = gcache.Remove(ctx, "paymentOrder_ListByStatus_failed")
@@ -115,7 +115,7 @@ func (d *paymentOrderDao) UpdateFail(ctx context.Context, id int64, notifyRaw st
func (d *paymentOrderDao) ListByUserAndType(ctx context.Context, userId int64, orderType string) ([]*entity.PaymentOrder, error) {
var list []*entity.PaymentOrder
err := g.DB().Model(consts.TableNamePaymentOrder).Ctx(ctx).
err := g.DB(consts.DbGroupFinance).Model(consts.TableNamePaymentOrder).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: "paymentOrder_ListByUserAndType_" + gconv.String(userId) + "_" + orderType}).
Where("user_id", userId).
Where("order_type", orderType).
@@ -124,7 +124,7 @@ func (d *paymentOrderDao) ListByUserAndType(ctx context.Context, userId int64, o
}
func (d *paymentOrderDao) ListPageByUser(ctx context.Context, userId int64, page, pageSize int) (res []*entity.PaymentOrder, total int, err error) {
m := g.DB().Model(consts.TableNamePaymentOrder).Ctx(ctx).Where("user_id", userId)
m := g.DB(consts.DbGroupFinance).Model(consts.TableNamePaymentOrder).Ctx(ctx).Where("user_id", userId)
if pageSize == -1 {
r, err := m.OrderDesc("id").All()
if err != nil {
shortdrama/dao/region_pricing_dao.go
+19 -19
View File
@@ -19,7 +19,7 @@ type regionPricingDao struct{}
func init() {
ctx := context.Background()
if _, err := g.DB().Exec(ctx, `CREATE TABLE IF NOT EXISTS `+consts.TableNameRegionPricing+` (
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, `CREATE TABLE IF NOT EXISTS `+consts.TableNameRegionPricing+` (
id INTEGER PRIMARY KEY AUTOINCREMENT,
province TEXT NOT NULL DEFAULT '',
region TEXT NOT NULL DEFAULT '',
@@ -31,33 +31,33 @@ func init() {
)`); err != nil {
g.Log().Warningf(ctx, "create region_pricing table failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "ALTER TABLE "+consts.TableNameRegionPricing+" ADD COLUMN province TEXT NOT NULL DEFAULT ''"); err != nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, "ALTER TABLE "+consts.TableNameRegionPricing+" ADD COLUMN province TEXT NOT NULL DEFAULT ''"); err != nil {
g.Log().Warningf(ctx, "add column province failed: %v", err)
}
// add max_customers column for existing databases
if _, err := g.DB().Exec(ctx, "ALTER TABLE "+consts.TableNameRegionPricing+" ADD COLUMN max_customers INTEGER NOT NULL DEFAULT 0"); err != nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, "ALTER TABLE "+consts.TableNameRegionPricing+" ADD COLUMN max_customers INTEGER NOT NULL DEFAULT 0"); err != nil {
g.Log().Warningf(ctx, "add column max_customers failed: %v", err)
}
// drop old indexes, recreate with new constraints
if _, err := g.DB().Exec(ctx, `DROP INDEX IF EXISTS idx_region_pricing_uniq`); err != nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, `DROP INDEX IF EXISTS idx_region_pricing_uniq`); err != nil {
g.Log().Warningf(ctx, "drop index idx_region_pricing_uniq failed: %v", err)
}
if _, err := g.DB().Exec(ctx, `DROP INDEX IF EXISTS idx_rp_price`); err != nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, `DROP INDEX IF EXISTS idx_rp_price`); err != nil {
g.Log().Warningf(ctx, "drop index idx_rp_price failed: %v", err)
}
if _, err := g.DB().Exec(ctx, `DROP INDEX IF EXISTS idx_rp_max_customers`); err != nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, `DROP INDEX IF EXISTS idx_rp_max_customers`); err != nil {
g.Log().Warningf(ctx, "drop index idx_rp_max_customers failed: %v", err)
}
if _, err := g.DB().Exec(ctx, `DROP INDEX IF EXISTS idx_rp_single_protected`); err != nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, `DROP INDEX IF EXISTS idx_rp_single_protected`); err != nil {
g.Log().Warningf(ctx, "drop index idx_rp_single_protected failed: %v", err)
}
if _, err := g.DB().Exec(ctx, `CREATE UNIQUE INDEX IF NOT EXISTS idx_rp_price ON `+consts.TableNameRegionPricing+`(region, protected, price)`); err != nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, `CREATE UNIQUE INDEX IF NOT EXISTS idx_rp_price ON `+consts.TableNameRegionPricing+`(region, protected, price)`); err != nil {
g.Log().Warningf(ctx, "create idx_rp_price failed: %v", err)
}
if _, err := g.DB().Exec(ctx, `CREATE UNIQUE INDEX IF NOT EXISTS idx_rp_max_customers ON `+consts.TableNameRegionPricing+`(region, protected, max_customers)`); err != nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, `CREATE UNIQUE INDEX IF NOT EXISTS idx_rp_max_customers ON `+consts.TableNameRegionPricing+`(region, protected, max_customers)`); err != nil {
g.Log().Warningf(ctx, "create idx_rp_max_customers failed: %v", err)
}
if _, err := g.DB().Exec(ctx, `CREATE UNIQUE INDEX IF NOT EXISTS idx_rp_single_protected ON `+consts.TableNameRegionPricing+`(region) WHERE protected=1`); err != nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, `CREATE UNIQUE INDEX IF NOT EXISTS idx_rp_single_protected ON `+consts.TableNameRegionPricing+`(region) WHERE protected=1`); err != nil {
g.Log().Warningf(ctx, "create idx_rp_single_protected failed: %v", err)
}
}
@@ -72,7 +72,7 @@ func clearRegionPricingCache(ctx context.Context) {
}
func (d *regionPricingDao) ListPage(ctx context.Context, page, pageSize int, keyword string, protected int) (res []*entity.RegionPricing, total int, err error) {
m := g.DB().Model(consts.TableNameRegionPricing).Ctx(ctx)
m := g.DB(consts.DbGroupSystem).Model(consts.TableNameRegionPricing).Ctx(ctx)
if keyword != "" {
m = m.Where("(region LIKE ? OR province LIKE ?)", "%"+keyword+"%", "%"+keyword+"%")
}
@@ -119,7 +119,7 @@ func (d *regionPricingDao) ListPage(ctx context.Context, page, pageSize int, key
func (d *regionPricingDao) List(ctx context.Context) ([]*entity.RegionPricing, error) {
var list []*entity.RegionPricing
err := g.DB().Model(consts.TableNameRegionPricing).Ctx(ctx).
err := g.DB(consts.DbGroupSystem).Model(consts.TableNameRegionPricing).Ctx(ctx).
Cache(gdb.CacheOption{Duration: time.Minute, Name: "region_pricing_List"}).
OrderAsc("region, protected").Scan(&list)
return list, err
@@ -127,7 +127,7 @@ func (d *regionPricingDao) List(ctx context.Context) ([]*entity.RegionPricing, e
func (d *regionPricingDao) GetOne(ctx context.Context, region string, protected int, maxCustomers int) (*entity.RegionPricing, error) {
var r entity.RegionPricing
err := g.DB().Model(consts.TableNameRegionPricing).Ctx(ctx).
err := g.DB(consts.DbGroupSystem).Model(consts.TableNameRegionPricing).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: "regionPricing_GetOne_" + region + "_" + gconv.String(protected) + "_" + gconv.String(maxCustomers)}).
Where("region", region).Where("protected", protected).Where("max_customers", maxCustomers).Scan(&r)
if err != nil || r.Id == 0 {
@@ -138,7 +138,7 @@ func (d *regionPricingDao) GetOne(ctx context.Context, region string, protected
func (d *regionPricingDao) GetByRegion(ctx context.Context, region string, protected int) (*entity.RegionPricing, error) {
var r entity.RegionPricing
err := g.DB().Model(consts.TableNameRegionPricing).Ctx(ctx).
err := g.DB(consts.DbGroupSystem).Model(consts.TableNameRegionPricing).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: "regionPricing_GetByRegion_" + region + "_" + gconv.String(protected)}).
Where("region", region).Where("protected", protected).Scan(&r)
if err != nil || r.Id == 0 {
@@ -149,7 +149,7 @@ func (d *regionPricingDao) GetByRegion(ctx context.Context, region string, prote
func (d *regionPricingDao) GetById(ctx context.Context, id int64) (*entity.RegionPricing, error) {
var r entity.RegionPricing
err := g.DB().Model(consts.TableNameRegionPricing).Ctx(ctx).
err := g.DB(consts.DbGroupSystem).Model(consts.TableNameRegionPricing).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: "regionPricing_GetById_" + gconv.String(id)}).
Where("id", id).Scan(&r)
if err != nil || r.Id == 0 {
@@ -160,13 +160,13 @@ func (d *regionPricingDao) GetById(ctx context.Context, id int64) (*entity.Regio
func (d *regionPricingDao) Save(ctx context.Context, data *entity.RegionPricing) error {
if data.Id > 0 {
_, err := g.DB().Exec(ctx,
_, err := g.DB(consts.DbGroupSystem).Exec(ctx,
"UPDATE "+consts.TableNameRegionPricing+" SET province=?, region=?, protected=?, price=?, max_customers=?, updated_at=datetime('now','localtime') WHERE id=?",
data.Province, data.Region, data.Protected, data.Price, data.MaxCustomers, data.Id)
clearRegionPricingCache(ctx)
return err
}
_, err := g.DB().Exec(ctx,
_, err := g.DB(consts.DbGroupSystem).Exec(ctx,
"INSERT INTO "+consts.TableNameRegionPricing+" (province, region, protected, price, max_customers, created_at, updated_at) VALUES (?, ?, ?, ?, ?, datetime('now','localtime'), datetime('now','localtime'))",
data.Province, data.Region, data.Protected, data.Price, data.MaxCustomers)
clearRegionPricingCache(ctx)
@@ -174,7 +174,7 @@ func (d *regionPricingDao) Save(ctx context.Context, data *entity.RegionPricing)
}
func (d *regionPricingDao) ListRegions(ctx context.Context) ([]string, error) {
values, err := g.DB().Model(consts.TableNameRegionPricing).Ctx(ctx).
values, err := g.DB(consts.DbGroupSystem).Model(consts.TableNameRegionPricing).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: "regionPricing_ListRegions"}).
Fields("DISTINCT region").OrderAsc("region").Array()
if err != nil {
@@ -188,7 +188,7 @@ func (d *regionPricingDao) ListRegions(ctx context.Context) ([]string, error) {
}
func (d *regionPricingDao) Delete(ctx context.Context, id int64) error {
_, err := g.DB().Model(consts.TableNameRegionPricing).Ctx(ctx).Where("id", id).Delete()
_, err := g.DB(consts.DbGroupSystem).Model(consts.TableNameRegionPricing).Ctx(ctx).Where("id", id).Delete()
clearRegionPricingCache(ctx)
return err
}
shortdrama/dao/user_dao.go
+55 -47
View File
@@ -18,7 +18,7 @@ type userDao struct{}
func init() {
ctx := context.Background()
_, err := g.DB().Exec(ctx, `CREATE TABLE IF NOT EXISTS `+consts.TableNameUser+` (
_, err := g.DB(consts.DbGroupFinance).Exec(ctx, `CREATE TABLE IF NOT EXISTS `+consts.TableNameUser+` (
id INTEGER PRIMARY KEY AUTOINCREMENT,
role TEXT NOT NULL DEFAULT 'customer',
username TEXT NOT NULL DEFAULT '',
@@ -33,73 +33,68 @@ func init() {
if err != nil {
g.Log().Warningf(ctx, "create user table failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "CREATE UNIQUE INDEX IF NOT EXISTS idx_user_username ON "+consts.TableNameUser+"(username) WHERE username != ''"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "CREATE UNIQUE INDEX IF NOT EXISTS idx_user_username ON "+consts.TableNameUser+"(username) WHERE username != ''"); err != nil {
g.Log().Warningf(ctx, "create index idx_user_username failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "CREATE UNIQUE INDEX IF NOT EXISTS idx_user_phone ON "+consts.TableNameUser+"(phone) WHERE phone != ''"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "CREATE UNIQUE INDEX IF NOT EXISTS idx_user_phone ON "+consts.TableNameUser+"(phone) WHERE phone != ''"); err != nil {
g.Log().Warningf(ctx, "create index idx_user_phone failed: %v", err)
}
// add columns for existing databases
if _, err := g.DB().Exec(ctx, "ALTER TABLE "+consts.TableNameUser+" ADD COLUMN province TEXT NOT NULL DEFAULT ''"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "ALTER TABLE "+consts.TableNameUser+" ADD COLUMN province TEXT NOT NULL DEFAULT ''"); err != nil {
g.Log().Warningf(ctx, "add column province failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "ALTER TABLE "+consts.TableNameUser+" ADD COLUMN address TEXT NOT NULL DEFAULT ''"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "ALTER TABLE "+consts.TableNameUser+" ADD COLUMN address TEXT NOT NULL DEFAULT ''"); err != nil {
g.Log().Warningf(ctx, "add column address failed: %v", err)
}
// cleanup legacy columns
if _, err := g.DB().Exec(ctx, "ALTER TABLE "+consts.TableNameUser+" DROP COLUMN status"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "ALTER TABLE "+consts.TableNameUser+" DROP COLUMN status"); err != nil {
g.Log().Warningf(ctx, "drop column status failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "ALTER TABLE "+consts.TableNameUser+" DROP COLUMN expired_at"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "ALTER TABLE "+consts.TableNameUser+" DROP COLUMN expired_at"); err != nil {
g.Log().Warningf(ctx, "drop column expired_at failed: %v", err)
}
// backfill created_at for legacy data
if _, err := g.DB().Exec(ctx, "UPDATE "+consts.TableNameUser+" SET created_at = datetime('now','localtime') WHERE created_at IS NULL"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "UPDATE "+consts.TableNameUser+" SET created_at = datetime('now','localtime') WHERE created_at IS NULL"); err != nil {
g.Log().Warningf(ctx, "backfill user created_at failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "UPDATE "+consts.TableNameAccountTransaction+" SET created_at = datetime('now','localtime') WHERE created_at IS NULL"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "UPDATE "+consts.TableNameAccountTransaction+" SET created_at = datetime('now','localtime') WHERE created_at IS NULL"); err != nil {
g.Log().Warningf(ctx, "backfill account_transaction created_at failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "UPDATE "+consts.TableNamePaymentOrder+" SET created_at = datetime('now','localtime') WHERE created_at IS NULL"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "UPDATE "+consts.TableNamePaymentOrder+" SET created_at = datetime('now','localtime') WHERE created_at IS NULL"); err != nil {
g.Log().Warningf(ctx, "backfill payment_order created_at failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "UPDATE "+consts.TableNamePaymentConfig+" SET created_at = datetime('now','localtime') WHERE created_at IS NULL"); err != nil {
g.Log().Warningf(ctx, "backfill payment_config created_at failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "UPDATE "+consts.TableNamePaymentChannelTrade+" SET created_at = datetime('now','localtime') WHERE created_at IS NULL"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "UPDATE "+consts.TableNamePaymentChannelTrade+" SET created_at = datetime('now','localtime') WHERE created_at IS NULL"); err != nil {
g.Log().Warningf(ctx, "backfill payment_channel_trade created_at failed: %v", err)
}
// UTC to local timezone migration, run once via PRAGMA user_version
var userVersion int
r, _ := g.DB().Query(ctx, "PRAGMA user_version")
if len(r) > 0 {
if r, qErr := g.DB(consts.DbGroupFinance).Query(ctx, "PRAGMA user_version"); qErr != nil {
g.Log().Warningf(ctx, "query user_version failed: %v", qErr)
} else if len(r) > 0 {
userVersion = r[0]["user_version"].Int()
}
if userVersion == 0 {
g.Log().Info(ctx, "migration: convert historical UTC time data to local timezone")
if _, err := g.DB().Exec(ctx, "UPDATE "+consts.TableNameUser+" SET created_at = datetime(created_at, '+8 hours'), updated_at = datetime(updated_at, '+8 hours') WHERE created_at LIKE '%-%T%:%.%Z' OR created_at LIKE '%-%T%:%s%z'"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "UPDATE "+consts.TableNameUser+" SET created_at = datetime(created_at, '+8 hours'), updated_at = datetime(updated_at, '+8 hours') WHERE created_at LIKE '%-%T%:%.%Z' OR created_at LIKE '%-%T%:%s%z'"); err != nil {
g.Log().Warningf(ctx, "migrate user UTC to local failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "UPDATE "+consts.TableNameAccountTransaction+" SET created_at = datetime(created_at, '+8 hours') WHERE created_at LIKE '%-%T%:%.%Z' OR created_at LIKE '%-%T%:%s%z' AND created_at IS NOT NULL"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "UPDATE "+consts.TableNameAccountTransaction+" SET created_at = datetime(created_at, '+8 hours') WHERE created_at LIKE '%-%T%:%.%Z' OR created_at LIKE '%-%T%:%s%z' AND created_at IS NOT NULL"); err != nil {
g.Log().Warningf(ctx, "migrate account_transaction UTC to local failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "UPDATE "+consts.TableNamePaymentOrder+" SET created_at = datetime(created_at, '+8 hours'), updated_at = datetime(updated_at, '+8 hours'), paid_at = datetime(paid_at, '+8 hours') WHERE created_at LIKE '%-%T%:%.%Z' OR created_at LIKE '%-%T%:%s%z'"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "UPDATE "+consts.TableNamePaymentOrder+" SET created_at = datetime(created_at, '+8 hours'), updated_at = datetime(updated_at, '+8 hours'), paid_at = datetime(paid_at, '+8 hours') WHERE created_at LIKE '%-%T%:%.%Z' OR created_at LIKE '%-%T%:%s%z'"); err != nil {
g.Log().Warningf(ctx, "migrate payment_order UTC to local failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "UPDATE "+consts.TableNamePaymentConfig+" SET created_at = datetime(created_at, '+8 hours') WHERE created_at LIKE '%-%T%:%.%Z' OR created_at LIKE '%-%T%:%s%z'"); err != nil {
g.Log().Warningf(ctx, "migrate payment_config UTC to local failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "UPDATE "+consts.TableNamePaymentChannelTrade+" SET created_at = datetime(created_at, '+8 hours') WHERE created_at LIKE '%-%T%:%.%Z' OR created_at LIKE '%-%T%:%s%z'"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "UPDATE "+consts.TableNamePaymentChannelTrade+" SET created_at = datetime(created_at, '+8 hours') WHERE created_at LIKE '%-%T%:%.%Z' OR created_at LIKE '%-%T%:%s%z'"); err != nil {
g.Log().Warningf(ctx, "migrate payment_channel_trade UTC to local failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "PRAGMA user_version = 1"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "PRAGMA user_version = 1"); err != nil {
g.Log().Warningf(ctx, "set user_version failed: %v", err)
}
}
// 补充 province 字段到各表的 DDL（仅针对在 province 列存在前入库的老数据）
if _, err := g.DB().Exec(ctx, "UPDATE "+consts.TableNameUser+" SET province = region WHERE province = '' AND region != ''"); err != nil {
if _, err := g.DB(consts.DbGroupFinance).Exec(ctx, "UPDATE "+consts.TableNameUser+" SET province = region WHERE province = '' AND region != ''"); err != nil {
g.Log().Warningf(ctx, "backfill province from region failed: %v", err)
}
}
@@ -113,7 +108,7 @@ func clearUserCache(ctx context.Context, id int64) {
}
func (d *userDao) Insert(ctx context.Context, data *entity.User) (int64, error) {
r, err := g.DB().Exec(ctx,
r, err := g.DB(consts.DbGroupFinance).Exec(ctx,
"INSERT INTO "+consts.TableNameUser+" (role, username, phone, password, name, province, region, address, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, datetime('now','localtime'), datetime('now','localtime'))",
data.Role, data.Username, data.Phone, data.Password, data.Name, data.Province, data.Region, data.Address)
if err != nil {
@@ -124,7 +119,7 @@ func (d *userDao) Insert(ctx context.Context, data *entity.User) (int64, error)
func (d *userDao) GetOne(ctx context.Context, id int64) (*entity.User, error) {
var u entity.User
err := g.DB().Model(consts.TableNameUser).Ctx(ctx).
err := g.DB(consts.DbGroupFinance).Model(consts.TableNameUser).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: "user_GetOne_" + gconv.String(id)}).
Where("id", id).Scan(&u)
if err != nil {
@@ -138,7 +133,7 @@ func (d *userDao) GetOne(ctx context.Context, id int64) (*entity.User, error) {
func (d *userDao) GetByUsername(ctx context.Context, username string) (*entity.User, error) {
var u entity.User
err := g.DB().Model(consts.TableNameUser).Ctx(ctx).
err := g.DB(consts.DbGroupFinance).Model(consts.TableNameUser).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: "user_GetByUsername_" + username}).
Where("username", username).Scan(&u)
if err != nil {
@@ -152,7 +147,7 @@ func (d *userDao) GetByUsername(ctx context.Context, username string) (*entity.U
func (d *userDao) GetByAccount(ctx context.Context, account string) (*entity.User, error) {
var u entity.User
err := g.DB().Model(consts.TableNameUser).Ctx(ctx).
err := g.DB(consts.DbGroupFinance).Model(consts.TableNameUser).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: "user_GetByAccount_" + account}).
Where("username = ? OR phone = ?", account, account).Scan(&u)
if err != nil {
@@ -166,7 +161,7 @@ func (d *userDao) GetByAccount(ctx context.Context, account string) (*entity.Use
func (d *userDao) GetByPhone(ctx context.Context, phone string) (*entity.User, error) {
var u entity.User
err := g.DB().Model(consts.TableNameUser).Ctx(ctx).
err := g.DB(consts.DbGroupFinance).Model(consts.TableNameUser).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: "user_GetByPhone_" + phone}).
Where("phone", phone).Scan(&u)
if err != nil || u.Id == 0 {
@@ -176,47 +171,60 @@ func (d *userDao) GetByPhone(ctx context.Context, phone string) (*entity.User, e
}
func (d *userDao) Update(ctx context.Context, data *entity.User) error {
_, err := g.DB().Model(consts.TableNameUser).Ctx(ctx).Data(data).Where("id", data.Id).Update()
_, err := g.DB(consts.DbGroupFinance).Model(consts.TableNameUser).Ctx(ctx).Data(data).Where("id", data.Id).Update()
clearUserCache(ctx, data.Id)
return err
}
func (d *userDao) UpdateFields(ctx context.Context, id int64, data g.Map) error {
_, err := g.DB().Model(consts.TableNameUser).Ctx(ctx).Data(data).Where("id", id).Update()
_, err := g.DB(consts.DbGroupFinance).Model(consts.TableNameUser).Ctx(ctx).Data(data).Where("id", id).Update()
clearUserCache(ctx, id)
return err
}
func (d *userDao) ListByRole(ctx context.Context, role string, page, pageSize int) ([]*entity.User, int, error) {
cacheKey := "user_ListByRole_" + role
total, err := g.DB().Model(consts.TableNameUser).Ctx(ctx).Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: cacheKey + "_count"}).Where("role", role).Count()
total, err := g.DB(consts.DbGroupFinance).Model(consts.TableNameUser).Ctx(ctx).Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: cacheKey + "_count"}).Where("role", role).Count()
if err != nil {
return nil, 0, err
}
var users []*entity.User
err = g.DB().Model(consts.TableNameUser).Ctx(ctx).Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: cacheKey}).Where("role", role).Page(page, pageSize).OrderAsc("id").Scan(&users)
err = g.DB(consts.DbGroupFinance).Model(consts.TableNameUser).Ctx(ctx).Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: cacheKey}).Where("role", role).Page(page, pageSize).OrderAsc("id").Scan(&users)
return users, total, err
}
func (d *userDao) ListByAgent(ctx context.Context, agentId int64, page, pageSize int) ([]*entity.User, int, error) {
if page < 1 {
page = 1
}
if pageSize < 1 {
pageSize = 20
}
cacheKey := "user_ListByAgent_" + gconv.String(agentId)
// 查询指定代理商名下的客户
total, err := g.DB().Model(consts.TableNameUser+" u").
InnerJoin(consts.TableNameCustomerProfile+" cp", "cp.user_id = u.id").
Where("cp.agent_id", agentId).
// 单表查询：客户档案表统计并分页取 user_id（customer_profile 主键即 user_id，顺序与 user 表一致）
total, err := g.DB(consts.DbGroupFinance).Model(consts.TableNameCustomerProfile).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: cacheKey + "_count"}).
Count()
Where("agent_id", agentId).Count()
if err != nil {
return nil, 0, err
}
var users []*entity.User
err = g.DB().Model(consts.TableNameUser+" u").
InnerJoin(consts.TableNameCustomerProfile+" cp", "cp.user_id = u.id").
Where("cp.agent_id", agentId).
r, err := g.DB(consts.DbGroupFinance).Model(consts.TableNameCustomerProfile).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: cacheKey}).
Fields("u.*").
Page(page, pageSize).
OrderAsc("u.id").
Scan(&users)
Fields("user_id").Where("agent_id", agentId).
OrderAsc("user_id").Limit(pageSize).Offset((page - 1) * pageSize).Array()
if err != nil {
return nil, 0, err
}
ids := make([]int64, 0, len(r))
for _, v := range r {
ids = append(ids, v.Int64())
}
if len(ids) == 0 {
return make([]*entity.User, 0), total, nil
}
// 批量查询用户信息
var users []*entity.User
err = g.DB(consts.DbGroupFinance).Model(consts.TableNameUser).Ctx(ctx).
WhereIn("id", ids).OrderAsc("id").Scan(&users)
return users, total, err
}
shortdrama/dao/user_model_config_dao.go
+25 -19
View File
@@ -24,14 +24,14 @@ func init() {
ctx := context.Background()
// 检测表是否已存在
var tableExists bool
if r, err := g.DB().Exec(ctx, "SELECT count(*) FROM sqlite_master WHERE type='table' AND name='"+consts.TableNameUserModelConfig+"'"); err != nil {
if r, err := g.DB(consts.DbGroupSystem).Exec(ctx, "SELECT count(*) FROM sqlite_master WHERE type='table' AND name='"+consts.TableNameUserModelConfig+"'"); err != nil {
g.Log().Warningf(ctx, "check user_model_config table existence failed: %v", err)
} else if r != nil {
tableExists = true
}
if !tableExists {
_, err := g.DB().Exec(ctx, `CREATE TABLE `+consts.TableNameUserModelConfig+` (
_, err := g.DB(consts.DbGroupSystem).Exec(ctx, `CREATE TABLE `+consts.TableNameUserModelConfig+` (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL DEFAULT 0,
model_config_id INTEGER NOT NULL DEFAULT 0,
@@ -53,7 +53,7 @@ func init() {
}
// 表已存在，检测是否包含旧列（base_url），有则迁移到不含 base_url/task_callback_url 的新表
r, err := g.DB().GetAll(ctx, "PRAGMA table_info("+consts.TableNameUserModelConfig+")")
r, err := g.DB(consts.DbGroupSystem).GetAll(ctx, "PRAGMA table_info("+consts.TableNameUserModelConfig+")")
var hasBaseUrl bool
if err != nil {
g.Log().Warningf(ctx, "check user_model_config table columns failed: %v", err)
@@ -68,7 +68,7 @@ func init() {
g.Log().Info(ctx, "检测到 user_model_config 表含旧列(base_url)，开始迁移到新结构...")
newTable := consts.TableNameUserModelConfig + "_new"
if _, err := g.DB().Exec(ctx, `CREATE TABLE `+newTable+` (
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, `CREATE TABLE `+newTable+` (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL DEFAULT 0,
model_config_id INTEGER NOT NULL DEFAULT 0,
@@ -88,7 +88,7 @@ func init() {
}
// 复制数据：旧表的 endpoint_url/interface_path/callback_path 已存在则原样保留，否则从 base_url 填充
if _, err := g.DB().Exec(ctx,
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx,
`INSERT INTO `+newTable+`(user_id,model_config_id,model_type,api_key,endpoint_url,interface_path,callback_path,temperature,max_tokens,concurrency_count,created_at,updated_at)
SELECT uc.user_id,uc.model_config_id,
COALESCE(mc.model_type,''),
@@ -106,10 +106,10 @@ func init() {
g.Log().Error(ctx, "迁移数据失败:", err)
}
if _, err := g.DB().Exec(ctx, "DROP TABLE "+consts.TableNameUserModelConfig); err != nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, "DROP TABLE "+consts.TableNameUserModelConfig); err != nil {
g.Log().Warningf(ctx, "drop old table failed: %v", err)
}
if _, err := g.DB().Exec(ctx, "ALTER TABLE "+newTable+" RENAME TO "+consts.TableNameUserModelConfig); err != nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, "ALTER TABLE "+newTable+" RENAME TO "+consts.TableNameUserModelConfig); err != nil {
g.Log().Error(ctx, "重命名表失败:", err)
return
}
@@ -117,7 +117,7 @@ func init() {
}
// 检测是否缺少 concurrency_count 列（已有新表结构但字段不全）
if r, err := g.DB().GetAll(ctx, "PRAGMA table_info("+consts.TableNameUserModelConfig+")"); err != nil {
if r, err := g.DB(consts.DbGroupSystem).GetAll(ctx, "PRAGMA table_info("+consts.TableNameUserModelConfig+")"); err != nil {
g.Log().Warningf(ctx, "check user_model_config concurrency_count column failed: %v", err)
} else if r != nil {
hasCol := false
@@ -129,7 +129,7 @@ func init() {
}
if !hasCol {
g.Log().Info(ctx, "检测到 user_model_config 缺少 concurrency_count 列，正在补充...")
if _, err := g.DB().Exec(ctx, "ALTER TABLE "+consts.TableNameUserModelConfig+" ADD COLUMN concurrency_count INTEGER NOT NULL DEFAULT 0"); err != nil {
if _, err := g.DB(consts.DbGroupSystem).Exec(ctx, "ALTER TABLE "+consts.TableNameUserModelConfig+" ADD COLUMN concurrency_count INTEGER NOT NULL DEFAULT 0"); err != nil {
g.Log().Warningf(ctx, "add concurrency_count column failed: %v", err)
} else {
g.Log().Info(ctx, "concurrency_count 列补充完成")
@@ -140,7 +140,7 @@ func init() {
// GetByUserId 获取指定用户所有模型配置（降序排列）
func (d *userModelConfigDao) GetByUserId(ctx context.Context, userId int64) (res []*entity.UserModelConfig, err error) {
r, err := g.DB().Model(consts.TableNameUserModelConfig).Ctx(ctx).
r, err := g.DB(consts.DbGroupSystem).Model(consts.TableNameUserModelConfig).Ctx(ctx).
Cache(gdb.CacheOption{Duration: time.Minute, Name: fmt.Sprintf("user_model_config_GetByUserId_%d", userId)}).
Where("user_id", userId).OrderDesc("id").All()
if err != nil {
@@ -156,7 +156,7 @@ func (d *userModelConfigDao) GetByUserId(ctx context.Context, userId int64) (res
// GetByModelConfigId 获取指定用户对指定系统模型的配置
func (d *userModelConfigDao) GetByModelConfigId(ctx context.Context, userId, modelConfigId int64) (res *entity.UserModelConfig, err error) {
r, err := g.DB().Model(consts.TableNameUserModelConfig).Ctx(ctx).
r, err := g.DB(consts.DbGroupSystem).Model(consts.TableNameUserModelConfig).Ctx(ctx).
Cache(gdb.CacheOption{Duration: common.CacheTTL(), Name: fmt.Sprintf("user_model_config_GetByModelConfigId_%d_%d", userId, modelConfigId)}).
Where("user_id", userId).Where("model_config_id", modelConfigId).Limit(1).One()
if err != nil {
@@ -172,7 +172,7 @@ func (d *userModelConfigDao) GetByModelConfigId(ctx context.Context, userId, mod
// GetByModelType 获取指定用户对指定模型类型的配置
func (d *userModelConfigDao) GetByModelType(ctx context.Context, userId int64, modelType string) (res *entity.UserModelConfig, err error) {
r, err := g.DB().Model(consts.TableNameUserModelConfig).Ctx(ctx).
r, err := g.DB(consts.DbGroupSystem).Model(consts.TableNameUserModelConfig).Ctx(ctx).
Cache(gdb.CacheOption{Duration: time.Minute, Name: fmt.Sprintf("user_model_config_GetByModelType_%d_%s", userId, modelType)}).
Where("user_id", userId).
Where("model_type", modelType).
@@ -191,22 +191,28 @@ func (d *userModelConfigDao) GetByModelType(ctx context.Context, userId int64, m
// GetFirstWithApiKeyByModelType 获取指定模型类型第一个有 API Key 的用户配置
func (d *userModelConfigDao) GetFirstWithApiKeyByModelType(ctx context.Context, modelType string) *entity.UserModelConfig {
r, err := g.DB().Model(consts.TableNameUserModelConfig).Ctx(ctx).
r, err := g.DB(consts.DbGroupSystem).Model(consts.TableNameUserModelConfig).Ctx(ctx).
Where("model_type", modelType).
Where("api_key != ''").
Limit(1).
One()
if err != nil || r == nil {
if err != nil {
g.Log().Warningf(ctx, "query user model config by type %s failed: %v", modelType, err)
return nil
}
if r == nil {
return nil
}
res := new(entity.UserModelConfig)
_ = r.Struct(&res)
if err := r.Struct(&res); err != nil {
g.Log().Warningf(ctx, "parse user model config row failed: %v", err)
}
return res
}
// DeleteSameType 删除同用户同模型类型的其他配置（保留当前配置）
func (d *userModelConfigDao) DeleteSameType(ctx context.Context, userId int64, modelType string, excludeId int64) error {
_, err := g.DB().Exec(ctx,
_, err := g.DB(consts.DbGroupSystem).Exec(ctx,
`DELETE FROM `+consts.TableNameUserModelConfig+`
WHERE user_id=? AND id!=? AND model_type=?`,
userId, excludeId, modelType)
@@ -221,12 +227,12 @@ func (d *userModelConfigDao) Save(ctx context.Context, data *entity.UserModelCon
now := gtime.Now().Format("Y-m-d H:i:s")
m["updated_at"] = now
if data.Id > 0 {
_, err := g.DB().Model(consts.TableNameUserModelConfig).Ctx(ctx).Data(m).Where("id", data.Id).Update()
_, err := g.DB(consts.DbGroupSystem).Model(consts.TableNameUserModelConfig).Ctx(ctx).Data(m).Where("id", data.Id).Update()
clearUserModelConfigCache(ctx, data.UserId, data.ModelType, data.ModelConfigId)
return err
}
m["created_at"] = now
lid, err := g.DB().Model(consts.TableNameUserModelConfig).Ctx(ctx).Data(m).InsertAndGetId()
lid, err := g.DB(consts.DbGroupSystem).Model(consts.TableNameUserModelConfig).Ctx(ctx).Data(m).InsertAndGetId()
if err == nil && lid > 0 {
data.Id = lid
}
@@ -238,7 +244,7 @@ func (d *userModelConfigDao) Save(ctx context.Context, data *entity.UserModelCon
// SaveBatch 批量保存用户模型配置（事务内执行）
func (d *userModelConfigDao) SaveBatch(ctx context.Context, items []*entity.UserModelConfig) error {
err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
err := g.DB(consts.DbGroupSystem).Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
for _, data := range items {
m := gconv.Map(data, gconv.MapOption{Tags: []string{"orm"}})
delete(m, "id")
shortdrama/model/agent_output.go
+12 -3
View File
@@ -2,8 +2,11 @@ package model
import (
"bytes"
"context"
"encoding/json"
"strings"
"github.com/gogf/gf/v2/frame/g"
)
// AgentOutput Agent 最终输出的 JSON 结构
@@ -91,13 +94,17 @@ func ParseAgentOutput(jsonStr string, segIdx int) *SegmentOutput {
if err := json.Unmarshal([]byte(extracted), &agentOut); err != nil {
// JSON 解析失败时尝试修复常见错误（如对话中的未转义引号）
if fixed := repairJSON(extracted); fixed != extracted {
_ = json.Unmarshal([]byte(fixed), &agentOut)
if fErr := json.Unmarshal([]byte(fixed), &agentOut); fErr != nil {
g.Log().Warningf(context.Background(), "parse agent output failed after repair: %v", fErr)
}
} else if len(extracted) > 0 && extracted[0] == '{' {
// 尝试修复整个 textOutput 中的 JSON
fixed2 := repairJSON(jsonStr)
extracted2 := extractJSON(fixed2)
if extracted2 != jsonStr {
_ = json.Unmarshal([]byte(extracted2), &agentOut)
if fErr := json.Unmarshal([]byte(extracted2), &agentOut); fErr != nil {
g.Log().Warningf(context.Background(), "parse agent output failed after full repair: %v", fErr)
}
}
}
if agentOut.Title == "" {
@@ -179,7 +186,9 @@ func ExtractAgentImages(jsonStr string) *AgentExtractedImages {
var agentOut AgentOutput
if err := json.Unmarshal([]byte(extracted), &agentOut); err != nil {
if fixed := repairJSON(extracted); fixed != extracted {
_ = json.Unmarshal([]byte(fixed), &agentOut)
if fErr := json.Unmarshal([]byte(fixed), &agentOut); fErr != nil {
g.Log().Warningf(context.Background(), "extract agent images failed after repair: %v", fErr)
}
}
}
if agentOut.Title == "" {
shortdrama/service/agent_service.go
+70 -18
View File
@@ -81,12 +81,18 @@ func (s *agentService) ExtractRegionFromAddress(ctx context.Context, address str
// CreateAgent 创建代理商
func (s *agentService) CreateAgent(ctx context.Context, username, password, phone, name, province, region string, regionProtected bool) (*entity.User, error) {
existing, _ := dao.User.GetByUsername(ctx, username)
existing, err := dao.User.GetByUsername(ctx, username)
if err != nil {
return nil, err
}
if existing != nil {
return nil, errors.New("username already exists")
}
if phone != "" {
existing, _ = dao.User.GetByPhone(ctx, phone)
existing, err = dao.User.GetByPhone(ctx, phone)
if err != nil {
return nil, err
}
if existing != nil {
return nil, errors.New("phone already in use")
}
@@ -98,13 +104,45 @@ func (s *agentService) CreateAgent(ctx context.Context, username, password, phon
RegionProtected int `orm:"region_protected"`
}
var activeAgents []*agentRow
_ = g.DB().Model(consts.TableNameUser+" u").
InnerJoin(consts.TableNameAgentProfile+" ap", "ap.user_id = u.id").
Where("u.role", "agent").
Where("u.region", region).
Where("(ap.expired_at IS NULL OR ap.expired_at > datetime('now','localtime'))").
Fields("u.id", "ap.region_protected").
Scan(&activeAgents)
// 拆库后改为单表查询 + 内存合并：先查未过期代理商的 user_id
var activeIds []int64
if apRows, err := g.DB(consts.DbGroupFinance).Model(consts.TableNameAgentProfile).Ctx(ctx).
Fields("user_id").
Where("(expired_at IS NULL OR expired_at > datetime('now','localtime'))").
All(); err == nil {
for _, row := range apRows {
activeIds = append(activeIds, row["user_id"].Int64())
}
}
// 再查该区域的代理商 user_id
var regionAgentIds []int64
if len(activeIds) > 0 {
var agentUsers []*entity.User
if err := g.DB(consts.DbGroupFinance).Model(consts.TableNameUser).Ctx(ctx).
Fields("id").
Where("role", "agent").
Where("region", region).
WhereIn("id", activeIds).
Scan(&agentUsers); err == nil {
for _, u := range agentUsers {
regionAgentIds = append(regionAgentIds, u.Id)
}
}
}
// 最后批量查 profile 组装 region_protected
if len(regionAgentIds) > 0 {
var profiles []*entity.AgentProfile
if err := g.DB(consts.DbGroupFinance).Model(consts.TableNameAgentProfile).Ctx(ctx).
WhereIn("user_id", regionAgentIds).Scan(&profiles); err == nil {
for _, ap := range profiles {
protected := 0
if ap.RegionProtected {
protected = 1
}
activeAgents = append(activeAgents, &agentRow{Id: ap.UserId, RegionProtected: protected})
}
}
}
if regionProtected {
if len(activeAgents) > 0 {
@@ -123,12 +161,18 @@ func (s *agentService) CreateAgent(ctx context.Context, username, password, phon
if regionProtected {
protectedVal = 1
}
pricing, _ := dao.RegionPricing.GetByRegion(ctx, region, protectedVal)
if pricing == nil {
return nil, errors.New("region pricing not configured")
pricing, err := dao.RegionPricing.GetByRegion(ctx, region, protectedVal)
if err != nil || pricing == nil {
if err == nil {
err = errors.New("region pricing not configured")
}
return nil, err
}
hash, _ := bcryptGenerate(password)
hash, err := bcryptGenerate(password)
if err != nil {
return nil, err
}
user := &entity.User{
Role: "agent",
Username: username,
@@ -139,7 +183,7 @@ func (s *agentService) CreateAgent(ctx context.Context, username, password, phon
Region: region,
}
if err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
if err := g.DB(consts.DbGroupFinance).Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
userId, e := dao.User.Insert(ctx, user)
if e != nil {
return e
@@ -236,19 +280,27 @@ func (s *agentService) UpdateAgent(ctx context.Context, req *dto.UpdateAgentReq)
}
}
profile, _ := dao.AgentProfile.Get(ctx, req.Id)
profile, err := dao.AgentProfile.Get(ctx, req.Id)
if err != nil {
return err
}
if profile == nil {
maxCustomers := 0
if pricing, _ := dao.RegionPricing.GetByRegion(ctx, req.Region, 0); pricing != nil {
pricing, pErr := dao.RegionPricing.GetByRegion(ctx, req.Region, 0)
if pErr != nil {
g.Log().Warningf(ctx, "get region pricing for agent %d failed: %v", req.Id, pErr)
} else if pricing != nil {
maxCustomers = pricing.MaxCustomers
}
_ = dao.AgentProfile.Save(ctx, &entity.AgentProfile{
if err := dao.AgentProfile.Save(ctx, &entity.AgentProfile{
UserId: req.Id,
MaxCustomers: maxCustomers,
Renewals: 0,
ExpiredAt: gtime.Now().AddDate(1, 0, 0),
RegionProtected: false,
})
}); err != nil {
return err
}
}
return nil
}
shortdrama/service/customer_service.go
+71 -37
View File
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"video-factory/shortdrama/consts"
"video-factory/shortdrama/dao"
"video-factory/shortdrama/model/dto"
@@ -19,7 +20,10 @@ type customerService struct{}
var CustomerService = new(customerService)
func (s *customerService) GetBalance(ctx context.Context, userId int64) int64 {
cp, _ := dao.CustomerProfile.Get(ctx, userId)
cp, err := dao.CustomerProfile.Get(ctx, userId)
if err != nil {
g.Log().Warningf(ctx, "get customer profile %d failed: %v", userId, err)
}
if cp == nil {
return 0
}
@@ -33,23 +37,35 @@ func (s *customerService) CreateCustomer(ctx context.Context, phone, name, addre
}
agent, err := dao.User.GetOne(ctx, agentId)
if err != nil || agent == nil || agent.Role != "agent" {
if err != nil {
return nil, err
}
if agent == nil || agent.Role != "agent" {
return nil, errors.New("agent not found")
}
ap, _ := dao.AgentProfile.Get(ctx, agentId)
if ap != nil && ap.ExpiredAt != nil && ap.ExpiredAt.Before(gtime.Now()) {
return nil, errors.New("agent has expired")
ap, err := dao.AgentProfile.Get(ctx, agentId)
if err != nil {
return nil, err
}
ap, _ = dao.AgentProfile.Get(ctx, agentId)
if ap != nil && ap.MaxCustomers > 0 {
count, _ := dao.CustomerProfile.CountByAgent(ctx, agentId)
if count >= ap.MaxCustomers {
return nil, fmt.Errorf("max customer limit reached (%d)", ap.MaxCustomers)
if ap != nil {
if ap.ExpiredAt != nil && ap.ExpiredAt.Before(gtime.Now()) {
return nil, errors.New("agent has expired")
}
if ap.MaxCustomers > 0 {
count, err := dao.CustomerProfile.CountByAgent(ctx, agentId)
if err != nil {
return nil, err
}
if count >= ap.MaxCustomers {
return nil, fmt.Errorf("max customer limit reached (%d)", ap.MaxCustomers)
}
}
}
existing, _ := dao.User.GetByPhone(ctx, phone)
existing, err := dao.User.GetByPhone(ctx, phone)
if err != nil {
return nil, err
}
if existing != nil {
return nil, errors.New("phone number already registered")
}
@@ -58,7 +74,10 @@ func (s *customerService) CreateCustomer(ctx context.Context, phone, name, addre
if len(defaultPwd) > 6 {
defaultPwd = defaultPwd[len(defaultPwd)-6:]
}
hash, _ := bcryptGenerate(defaultPwd)
hash, err := bcryptGenerate(defaultPwd)
if err != nil {
return nil, err
}
user := &entity.User{
Role: "customer",
@@ -70,7 +89,7 @@ func (s *customerService) CreateCustomer(ctx context.Context, phone, name, addre
Address: address,
}
err = g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
err = g.DB(consts.DbGroupFinance).Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
userId, e := dao.User.Insert(ctx, user)
if e != nil {
return e
@@ -87,7 +106,10 @@ func (s *customerService) CreateCustomer(ctx context.Context, phone, name, addre
// CheckBalance 校验余额是否充足，返回 (是否充足, 预估费用(分))
func (s *customerService) CheckBalance(ctx context.Context, customerId int64, durationSec int64) (bool, int64, error) {
cp, _ := dao.CustomerProfile.Get(ctx, customerId)
cp, err := dao.CustomerProfile.Get(ctx, customerId)
if err != nil {
return false, 0, err
}
if cp == nil {
return false, 0, errors.New("customer not found")
}
@@ -96,29 +118,34 @@ func (s *customerService) CheckBalance(ctx context.Context, customerId int64, du
return cp.Balance >= cost, cost, nil
}
// DeductBalance 扣费
// DeductBalance 扣费：余额扣减与流水写入放在同一事务内，保证原子性
func (s *customerService) DeductBalance(ctx context.Context, customerId int64, amount int64, remark, createdBy string) error {
cp, err := dao.CustomerProfile.Get(ctx, customerId)
if err != nil || cp == nil {
return errors.New("customer not found")
}
if cp.Balance < amount {
return errors.New("insufficient balance")
}
newBalance := cp.Balance - amount
if err := dao.CustomerProfile.UpdateBalance(ctx, customerId, newBalance); err != nil {
return g.DB(consts.DbGroupFinance).Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
cp, err := dao.CustomerProfile.Get(ctx, customerId)
if err != nil {
return err
}
if cp == nil {
return errors.New("customer not found")
}
if cp.Balance < amount {
return errors.New("insufficient balance")
}
newBalance := cp.Balance - amount
if err := dao.CustomerProfile.UpdateBalance(ctx, customerId, newBalance); err != nil {
return err
}
_, err = dao.AccountTransaction.Insert(ctx, &entity.AccountTransaction{
UserId: customerId,
Type: "deduct",
Amount: amount,
BalanceBefore: cp.Balance,
BalanceAfter: newBalance,
Remark: remark,
CreatedBy: createdBy,
})
return err
}
_, err = dao.AccountTransaction.Insert(ctx, &entity.AccountTransaction{
UserId: customerId,
Type: "deduct",
Amount: amount,
BalanceBefore: cp.Balance,
BalanceAfter: newBalance,
Remark: remark,
CreatedBy: createdBy,
})
return err
}
// ListCustomers 分页查询客户列表
@@ -155,14 +182,21 @@ func (s *customerService) GetCustomerDetail(ctx context.Context, id int64) (*dto
if err != nil || user == nil {
return nil, err
}
profile, _ := dao.CustomerProfile.Get(ctx, id)
profile, err := dao.CustomerProfile.Get(ctx, id)
if err != nil {
g.Log().Warningf(ctx, "get customer profile %d failed: %v", id, err)
}
balance := s.GetBalance(ctx, id)
return &dto.CustomerDetail{User: user, Profile: profile, Balance: balance}, nil
}
// UpdateCustomer 更新客户信息
func (s *customerService) UpdateCustomer(ctx context.Context, agentId int64, req *dto.UpdateCustomerReq) error {
if ap, _ := dao.AgentProfile.Get(ctx, agentId); ap != nil && ap.ExpiredAt != nil && ap.ExpiredAt.Before(gtime.Now()) {
ap, err := dao.AgentProfile.Get(ctx, agentId)
if err != nil {
return err
}
if ap != nil && ap.ExpiredAt != nil && ap.ExpiredAt.Before(gtime.Now()) {
return errors.New("agent has expired, cannot edit customer")
}
shortdrama/service/drama_service.go
+8 -2
View File
@@ -23,7 +23,10 @@ var DramaService = new(dramaService)
// ==================== Drama CRUD ====================
func (s *dramaService) Create(ctx context.Context, title, contentType, config, aspectRatio string, episodeDuration int64, resolution string, userId int64) (int64, error) {
existing, _ := dao.Drama.GetByTitle(ctx, title)
existing, err := dao.Drama.GetByTitle(ctx, title)
if err != nil {
return 0, err
}
if existing != nil {
return 0, fmt.Errorf("drama title already exists: %s", title)
}
@@ -102,7 +105,10 @@ func (s *dramaService) Update(ctx context.Context, id int64, title, contentType,
}
oldTitle := d.Title
if title != "" && title != oldTitle {
existing, _ := dao.Drama.GetByTitle(ctx, title)
existing, err := dao.Drama.GetByTitle(ctx, title)
if err != nil {
return err
}
if existing != nil {
return fmt.Errorf("drama title already exists: %s", title)
}
shortdrama/service/episode_service.go
+63 -17
View File
@@ -185,7 +185,10 @@ func (s *episodeService) DeleteEpisode(ctx context.Context, dramaId, epId int64)
if !e.IsDir() {
name := e.Name()
if strings.HasPrefix(name, segPrefix) || name == safeEp+".mp4" {
_ = os.Remove(filepath.Join(videoDir, name))
fp := filepath.Join(videoDir, name)
if rmErr := os.Remove(fp); rmErr != nil && !os.IsNotExist(rmErr) {
g.Log().Warningf(ctx, "remove old video file %s failed: %v", fp, rmErr)
}
}
}
}
@@ -872,7 +875,10 @@ func createPendingTasks(ctx context.Context, dramaId, epId int64, script string,
// 只包含已有真实文件的实体（PortraitPath/ImagePath 非空），
// 全部加载（不设上限），上限在按段过滤后施加。
loadEntityRefs := func() []*entity.Character {
chars, _, _ := dao.Character.ListPageByDrama(ctx, dramaId, 1, -1)
chars, _, le := dao.Character.ListPageByDrama(ctx, dramaId, 1, -1)
if le != nil {
g.Log().Warningf(ctx, "load characters for refs failed: %v", le)
}
namedRefs = nil
for _, ch := range chars {
if ch.PortraitPath == "" {
@@ -880,14 +886,20 @@ func createPendingTasks(ctx context.Context, dramaId, epId int64, script string,
}
namedRefs = append(namedRefs, _namedRef{path: ch.PortraitPath, name: ch.Name})
}
scenes, _ := dao.Scene.ListByDrama(ctx, dramaId)
scenes, le := dao.Scene.ListByDrama(ctx, dramaId)
if le != nil {
g.Log().Warningf(ctx, "load scenes for refs failed: %v", le)
}
for _, sc := range scenes {
if sc.ImagePath == "" {
continue
}
namedRefs = append(namedRefs, _namedRef{path: sc.ImagePath, name: sc.Name})
}
props, _ := dao.Prop.ListByDrama(ctx, dramaId)
props, le := dao.Prop.ListByDrama(ctx, dramaId)
if le != nil {
g.Log().Warningf(ctx, "load props for refs failed: %v", le)
}
for _, p := range props {
if p.ImagePath == "" {
continue
@@ -962,7 +974,9 @@ func createPendingTasks(ctx context.Context, dramaId, epId int64, script string,
}
// ============ 从镜头脚本提取演员/场景/道具，自动写入数据库（避免重复）============
extractAndSaveEntities(ctx, dramaId, allShots)
if e := extractAndSaveEntities(ctx, dramaId, allShots); e != nil {
return e
}
// 重新加载实体引用，确保新增记录也被纳入后续的 media/prompt 构建
chars = loadEntityRefs()
@@ -1185,7 +1199,10 @@ func createPendingTasks(ctx context.Context, dramaId, epId int64, script string,
}
injectFirstFrame(body, "", modelCfg.SchemaMapping, modelCfg.Schema)
bodyJSON, _ := json.Marshal(body)
bodyJSON, mErr := json.Marshal(body)
if mErr != nil {
return fmt.Errorf("marshal segment %d body failed: %w", i, mErr)
}
records = append(records, g.Map{
"drama_id": dramaId,
@@ -1241,9 +1258,9 @@ func buildCharacterGuide(charNames []string, chars []*entity.Character, labelOf
// extractAndSaveEntities 从镜头数组中提取演员/场景/道具，去重后写入数据库。
// 如果同 drama 下已存在同名记录，跳过插入（避免重复）。
func extractAndSaveEntities(ctx context.Context, dramaId int64, shots []domain.Shot) {
func extractAndSaveEntities(ctx context.Context, dramaId int64, shots []domain.Shot) error {
if len(shots) == 0 {
return
return nil
}
charSet := make(map[string]bool)
@@ -1269,9 +1286,18 @@ func extractAndSaveEntities(ctx context.Context, dramaId int64, shots []domain.S
}
// 查询已有记录
existingChars, _, _ := dao.Character.ListPageByDrama(ctx, dramaId, 1, -1)
existingScenes, _ := dao.Scene.ListByDrama(ctx, dramaId)
existingProps, _ := dao.Prop.ListByDrama(ctx, dramaId)
existingChars, _, err := dao.Character.ListPageByDrama(ctx, dramaId, 1, -1)
if err != nil {
return fmt.Errorf("list existing characters failed: %w", err)
}
existingScenes, err := dao.Scene.ListByDrama(ctx, dramaId)
if err != nil {
return fmt.Errorf("list existing scenes failed: %w", err)
}
existingProps, err := dao.Prop.ListByDrama(ctx, dramaId)
if err != nil {
return fmt.Errorf("list existing props failed: %w", err)
}
existName := func(list interface{}, name string) bool {
switch l := list.(type) {
@@ -1299,41 +1325,61 @@ func extractAndSaveEntities(ctx context.Context, dramaId int64, shots []domain.S
now := time.Now().Format("2006-01-02 15:04:05")
// 收集新增记录后批量插入（避免循环单条查询/插入）
newChars := make([]g.Map, 0, len(charSet))
for name := range charSet {
if existName(existingChars, name) {
continue
}
_, _ = g.DB().Model(consts.TableNameCharacter).Ctx(ctx).Data(g.Map{
newChars = append(newChars, g.Map{
"drama_id": dramaId,
"name": name,
"created_at": now,
"updated_at": now,
}).Insert()
})
}
if len(newChars) > 0 {
if _, e := g.DB().Model(consts.TableNameCharacter).Ctx(ctx).Data(newChars).Insert(); e != nil {
return fmt.Errorf("batch insert characters failed: %w", e)
}
}
newScenes := make([]g.Map, 0, len(sceneSet))
for name := range sceneSet {
if existName(existingScenes, name) {
continue
}
_, _ = g.DB().Model(consts.TableNameScene).Ctx(ctx).Data(g.Map{
newScenes = append(newScenes, g.Map{
"drama_id": dramaId,
"name": name,
"created_at": now,
"updated_at": now,
}).Insert()
})
}
if len(newScenes) > 0 {
if _, e := g.DB().Model(consts.TableNameScene).Ctx(ctx).Data(newScenes).Insert(); e != nil {
return fmt.Errorf("batch insert scenes failed: %w", e)
}
}
newProps := make([]g.Map, 0, len(propSet))
for name := range propSet {
if existName(existingProps, name) {
continue
}
_, _ = g.DB().Model(consts.TableNameProp).Ctx(ctx).Data(g.Map{
newProps = append(newProps, g.Map{
"drama_id": dramaId,
"name": name,
"created_at": now,
"updated_at": now,
}).Insert()
})
}
if len(newProps) > 0 {
if _, e := g.DB().Model(consts.TableNameProp).Ctx(ctx).Data(newProps).Insert(); e != nil {
return fmt.Errorf("batch insert props failed: %w", e)
}
}
return nil
}
// inferCharPerSecond 根据台词情绪推断语速（字/秒）
shortdrama/service/generation_service.go
+113 -40
View File
@@ -76,7 +76,9 @@ func (s *generationService) GenerateEpisode(ctx context.Context, dramaId, epId i
// ============ 检查待生成任务的 media 文件是否存在 ============
if !continueIfMissing {
if missingNames := checkTaskMediaExistence(ctx, epId); len(missingNames) > 0 {
_ = dao.Episode.UpdateStatus(ctx, epId, consts.EpisodeStatusPending, "")
if e := dao.Episode.UpdateStatus(ctx, epId, consts.EpisodeStatusPending, ""); e != nil {
g.Log().Errorf(ctx, "reset episode %d status to pending failed: %v", epId, e)
}
return errors.New("至少需要包含一个人物形象")
}
} else {
@@ -161,8 +163,12 @@ func (s *generationService) GenerateEpisode(ctx context.Context, dramaId, epId i
g.Log().Infof(genCtx, "第%d集第%d段开始串行生成(segDur=%ds))", ep.Index, i+1, segDur)
if err := s.generateOneSegment(genCtx, d, ep, taskId, i, segDur, "", genCtx2); err != nil {
g.Log().Errorf(genCtx, "episode %d segment %d serial generation failed: %v", ep.Index, i+1, err)
_ = dao.GenerationTask.UpdateFailed(genCtx, taskId, err.Error())
_ = dao.Episode.UpdateStatus(ctx, epId, consts.EpisodeStatusPending, "")
if ufErr := dao.GenerationTask.UpdateFailed(genCtx, taskId, err.Error()); ufErr != nil {
g.Log().Errorf(genCtx, "mark task %d failed failed: %v", taskId, ufErr)
}
if ueErr := dao.Episode.UpdateStatus(ctx, epId, consts.EpisodeStatusPending, ""); ueErr != nil {
g.Log().Errorf(genCtx, "reset episode %d status to pending failed: %v", epId, ueErr)
}
clearPollCache(genCtx, epId)
return err
}
@@ -190,7 +196,9 @@ func (s *generationService) GenerateEpisode(ctx context.Context, dramaId, epId i
if r := recover(); r != nil {
errMsg := fmt.Sprintf("panic: %v", r)
g.Log().Errorf(genCtx, "episode %d segment %d generation panic: %v", ep.Index, idx+1, r)
_ = dao.GenerationTask.UpdateFailed(genCtx, tid, errMsg)
if ufErr := dao.GenerationTask.UpdateFailed(genCtx, tid, errMsg); ufErr != nil {
g.Log().Errorf(genCtx, "mark task %d failed after panic failed: %v", tid, ufErr)
}
clearPollCache(genCtx, epId)
}
<-sem
@@ -200,7 +208,9 @@ func (s *generationService) GenerateEpisode(ctx context.Context, dramaId, epId i
g.Log().Infof(genCtx, "第%d集第%d段开始生成(segDur=%ds)", ep.Index, idx+1, dur)
if err := s.generateOneSegment(genCtx, d, ep, tid, idx, dur, "", genCtx2); err != nil {
g.Log().Errorf(genCtx, "episode %d segment %d generation failed (elapsed %v): %v", ep.Index, idx+1, time.Since(startTime), err)
_ = dao.GenerationTask.UpdateFailed(genCtx, tid, err.Error())
if ufErr := dao.GenerationTask.UpdateFailed(genCtx, tid, err.Error()); ufErr != nil {
g.Log().Errorf(genCtx, "mark task %d failed failed: %v", tid, ufErr)
}
clearPollCache(genCtx, epId)
} else {
g.Log().Infof(genCtx, "第%d集第%d段生成完成(耗时%v)", ep.Index, idx+1, time.Since(startTime))
@@ -372,7 +382,9 @@ func (s *generationService) generateOneSegment(ctx context.Context, d *entity.Dr
updateFields["video_task_id"] = taskID
}
updateFields["model_name"] = videoCfg.ModelName
_, _ = g.DB().Model(consts.TableNameGenerationTask).Ctx(ctx).Data(updateFields).Where("id", taskId).Update()
if _, ue := g.DB().Model(consts.TableNameGenerationTask).Ctx(ctx).Data(updateFields).Where("id", taskId).Update(); ue != nil {
g.Log().Errorf(ctx, "update task %d after video submit failed: %v", taskId, ue)
}
return nil
}
@@ -729,7 +741,9 @@ func (s *generationService) FeedbackSegment(ctx context.Context, taskId int64, f
}
}
// 清除 DB 中的 video_url，防止轮询器使用旧数据
_ = dao.GenerationTask.UpdateFields(genCtx, taskId, g.Map{"video_url": "", "video_task_id": ""})
if e := dao.GenerationTask.UpdateFields(genCtx, taskId, g.Map{"video_url": "", "video_task_id": ""}); e != nil {
g.Log().Errorf(genCtx, "clear video_url of task %d failed: %v", taskId, e)
}
g.Log().Infof(genCtx, "第%d集第%d段重新提交视频(跳过Agent)", ep.Index, task.SegmentIdx+1)
@@ -755,25 +769,33 @@ func (s *generationService) FeedbackSegment(ctx context.Context, taskId int64, f
}
// 调用模型前写入模型名称
_ = dao.GenerationTask.UpdateFields(genCtx, taskId, g.Map{"model_name": modelCfg.ModelName})
if e := dao.GenerationTask.UpdateFields(genCtx, taskId, g.Map{"model_name": modelCfg.ModelName}); e != nil {
g.Log().Errorf(genCtx, "write model_name to task %d failed: %v", taskId, e)
}
newTaskID, resolvedBody, submitErr := resubmitVideoTask(genCtx, modelCfg.ApiKey, modelCfg.BaseUrl, modelCfg.Schema, bodyBytes, modelCfg.SchemaMapping)
if submitErr != nil {
g.Log().Errorf(genCtx, "episode %d segment %d video resubmit failed: %v", ep.Index, task.SegmentIdx+1, submitErr)
_ = dao.GenerationTask.UpdateFailed(genCtx, taskId, submitErr.Error())
if ufErr := dao.GenerationTask.UpdateFailed(genCtx, taskId, submitErr.Error()); ufErr != nil {
g.Log().Errorf(genCtx, "mark task %d failed failed: %v", taskId, ufErr)
}
clearPollCache(genCtx, task.EpisodeId)
return
}
g.Log().Infof(genCtx, "第%d集第%d段重提提交成功(taskId=%s)", ep.Index, task.SegmentIdx+1, newTaskID)
_ = dao.GenerationTask.UpdateFields(genCtx, taskId, g.Map{
if e := dao.GenerationTask.UpdateFields(genCtx, taskId, g.Map{
"video_task_id": newTaskID,
"script": string(resolvedBody),
"model_name": modelCfg.ModelName,
"updated_at": nil,
})
}); e != nil {
g.Log().Errorf(genCtx, "persist resubmit result to task %d failed: %v", taskId, e)
}
clearPollCache(genCtx, task.EpisodeId)
_ = dao.Episode.UpdateStatus(genCtx, task.EpisodeId, consts.EpisodeStatusGenerating, "")
if e := dao.Episode.UpdateStatus(genCtx, task.EpisodeId, consts.EpisodeStatusGenerating, ""); e != nil {
g.Log().Errorf(genCtx, "mark episode %d generating failed: %v", task.EpisodeId, e)
}
if ts, e := dao.GenerationTask.ListByEpisode(genCtx, task.EpisodeId); e == nil {
setPollCache(genCtx, task.EpisodeId, ts)
}
@@ -936,7 +958,10 @@ func cleanupEpisodeWorkspace(ctx context.Context, dramaTitle string, epIndex int
if !e.IsDir() {
name := e.Name()
if strings.HasPrefix(name, segPrefix) || name == safeEp+".mp4" || strings.HasPrefix(name, "concat_") {
_ = os.Remove(filepath.Join(videoDir, name))
fp := filepath.Join(videoDir, name)
if rmErr := os.Remove(fp); rmErr != nil && !os.IsNotExist(rmErr) {
g.Log().Warningf(ctx, "remove old video file %s failed: %v", fp, rmErr)
}
}
}
}
@@ -1585,12 +1610,14 @@ func setPollCache(ctx context.Context, epId int64, tasks []*entity.GenerationTas
if allDone {
status = consts.EpisodeStatusCompleted
}
_ = gcache.Set(ctx, fmt.Sprintf("episode_poll:%d", epId), &dto.EpisodePollRes{
if e := gcache.Set(ctx, fmt.Sprintf("episode_poll:%d", epId), &dto.EpisodePollRes{
Status: status,
ErrorMessage: errMsg,
Tasks: tasks,
CurrentTaskId: currentTaskId,
}, pollCacheTTL)
}, pollCacheTTL); e != nil {
g.Log().Warningf(ctx, "set poll cache for episode %d failed: %v", epId, e)
}
}
// clearPollCache 清除剧集轮询缓存
@@ -1612,7 +1639,9 @@ func (s *generationService) GetEpisodePollStatus(ctx context.Context, epId int64
emptyRes := &dto.EpisodePollRes{
Status: "pending",
}
_ = gcache.Set(ctx, cacheKey, emptyRes, 5*time.Second)
if cErr := gcache.Set(ctx, cacheKey, emptyRes, 5*time.Second); cErr != nil {
g.Log().Warningf(ctx, "set empty poll cache for episode %d failed: %v", epId, cErr)
}
return nil, nil
}
setPollCache(ctx, epId, tasks)
@@ -1652,7 +1681,11 @@ func (s *generationService) StartVideoPoller(ctx context.Context) {
// pollPendingVideos 扫描所有 generating 任务，按剧集分组推进（不同剧集并行，同剧集串行）
func (s *generationService) pollPendingVideos(ctx context.Context) {
allTasks, err := dao.GenerationTask.ListByStatuses(ctx, []string{consts.TaskStatusGenerating, consts.TaskStatusReview})
if err != nil || len(allTasks) == 0 {
if err != nil {
g.Log().Warningf(ctx, "poller: list generating tasks failed: %v", err)
return
}
if len(allTasks) == 0 {
return
}
@@ -1729,9 +1762,9 @@ func (s *generationService) processOneEpisode(ctx context.Context, tasks []*enti
}
// 加载短剧和 merged config
drama, _ := dao.Drama.GetOne(ctx, current.DramaId)
if drama == nil {
g.Log().Warningf(ctx, "poller: drama %d not found, skipping", current.DramaId)
drama, err := dao.Drama.GetOne(ctx, current.DramaId)
if err != nil || drama == nil {
g.Log().Warningf(ctx, "poller: load drama %d failed: %v", current.DramaId, err)
return
}
modelCfg := UserModelConfigService.GetMergedConfig(ctx, drama.UserId, "video")
@@ -1746,7 +1779,9 @@ func (s *generationService) processOneEpisode(ctx context.Context, tasks []*enti
if err != nil {
if strings.Contains(err.Error(), "FAILED") || strings.Contains(err.Error(), "failed") {
g.Log().Warningf(ctx, "poller: task %d segment %d video task failed: %v", current.Id, current.SegmentIdx+1, err)
_ = dao.GenerationTask.UpdateFailed(ctx, current.Id, err.Error())
if ufErr := dao.GenerationTask.UpdateFailed(ctx, current.Id, err.Error()); ufErr != nil {
g.Log().Errorf(ctx, "poller: mark task %d failed failed: %v", current.Id, ufErr)
}
clearPollCache(ctx, current.EpisodeId)
} else if strings.Contains(err.Error(), "RUNNING") {
g.Log().Debugf(ctx, "轮询器: 任务 %d 第%d段视频正在生成中，继续等待...", current.Id, current.SegmentIdx+1)
@@ -1770,16 +1805,24 @@ func (s *generationService) processOneEpisode(ctx context.Context, tasks []*enti
fmt.Sprintf("%s_seg_%d.mp4", safeEp, current.SegmentIdx))
if dlErr != nil {
g.Log().Warningf(ctx, "poller: task %d segment %d video download failed, using remote url: %v", current.Id, current.SegmentIdx+1, dlErr)
_ = dao.GenerationTask.UpdateFields(ctx, current.Id, g.Map{"video_url": videoURL, "video_task_id": ""})
if ue := dao.GenerationTask.UpdateFields(ctx, current.Id, g.Map{"video_url": videoURL, "video_task_id": ""}); ue != nil {
g.Log().Errorf(ctx, "poller: set remote video_url of task %d failed: %v", current.Id, ue)
}
} else {
g.Log().Infof(ctx, "轮询器: 任务 %d 第%d段视频已下载: %s", current.Id, current.SegmentIdx+1, localPath)
_ = dao.GenerationTask.UpdateFields(ctx, current.Id, g.Map{"video_url": localPath, "video_task_id": ""})
if ue := dao.GenerationTask.UpdateFields(ctx, current.Id, g.Map{"video_url": localPath, "video_task_id": ""}); ue != nil {
g.Log().Errorf(ctx, "poller: set local video_url of task %d failed: %v", current.Id, ue)
}
}
if ue := dao.GenerationTask.UpdateStatus(ctx, current.Id, consts.TaskStatusReview); ue != nil {
g.Log().Errorf(ctx, "poller: mark task %d review failed: %v", current.Id, ue)
}
_ = dao.GenerationTask.UpdateStatus(ctx, current.Id, consts.TaskStatusReview)
g.Log().Infof(ctx, "轮询器: 任务 %d 第%d段视频生成完成，进入审核", current.Id, current.SegmentIdx+1)
// 更新剧集状态：如果全部任务都到了 review/completed，剧集标记为 review
if allTasksDone, _ := dao.GenerationTask.ListByEpisode(ctx, current.EpisodeId); len(allTasksDone) > 0 {
if allTasksDone, leErr := dao.GenerationTask.ListByEpisode(ctx, current.EpisodeId); leErr != nil {
g.Log().Warningf(ctx, "poller: list tasks of episode %d failed: %v", current.EpisodeId, leErr)
} else if len(allTasksDone) > 0 {
allReview := true
for _, t := range allTasksDone {
if t.Status != consts.TaskStatusReview && t.Status != consts.TaskStatusCompleted {
@@ -1788,7 +1831,9 @@ func (s *generationService) processOneEpisode(ctx context.Context, tasks []*enti
}
}
if allReview {
_ = dao.Episode.UpdateStatus(ctx, current.EpisodeId, consts.EpisodeStatusReview, "")
if ue := dao.Episode.UpdateStatus(ctx, current.EpisodeId, consts.EpisodeStatusReview, ""); ue != nil {
g.Log().Errorf(ctx, "poller: mark episode %d review failed: %v", current.EpisodeId, ue)
}
}
}
return
@@ -1800,7 +1845,9 @@ func (s *generationService) processOneEpisode(ctx context.Context, tasks []*enti
var bodyMap map[string]any
if err := json.Unmarshal([]byte(current.Script), &bodyMap); err != nil {
g.Log().Errorf(ctx, "poller: task %d script is not valid JSON: %v", current.Id, err)
_ = dao.GenerationTask.UpdateFailed(ctx, current.Id, "script 不是合法 JSON")
if ufErr := dao.GenerationTask.UpdateFailed(ctx, current.Id, "script 不是合法 JSON"); ufErr != nil {
g.Log().Errorf(ctx, "poller: mark task %d failed failed: %v", current.Id, ufErr)
}
return
}
@@ -1837,7 +1884,9 @@ func (s *generationService) processOneEpisode(ctx context.Context, tasks []*enti
}
// 调用模型前写入模型名称
_ = dao.GenerationTask.UpdateFields(ctx, current.Id, g.Map{"model_name": modelCfg.ModelName})
if ue := dao.GenerationTask.UpdateFields(ctx, current.Id, g.Map{"model_name": modelCfg.ModelName}); ue != nil {
g.Log().Errorf(ctx, "poller: write model_name to task %d failed: %v", current.Id, ue)
}
taskId, _, err := resubmitVideoTask(ctx, modelCfg.ApiKey, modelCfg.BaseUrl, modelCfg.Schema, bodyJSON, modelCfg.SchemaMapping)
if err != nil {
@@ -1845,9 +1894,11 @@ func (s *generationService) processOneEpisode(ctx context.Context, tasks []*enti
return
}
_ = dao.GenerationTask.UpdateFields(ctx, current.Id, g.Map{
if ue := dao.GenerationTask.UpdateFields(ctx, current.Id, g.Map{
"video_task_id": taskId,
})
}); ue != nil {
g.Log().Errorf(ctx, "poller: persist video_task_id to task %d failed: %v", current.Id, ue)
}
g.Log().Infof(ctx, "轮询器: 第%d段视频已提交(taskId=%s)", current.SegmentIdx+1, taskId)
}
@@ -2158,7 +2209,10 @@ func callVideoAPI(ctx context.Context, apiKey, baseURL string, payload []byte, s
}
defer resp.Body.Close()
respData, _ := io.ReadAll(resp.Body)
respData, err := io.ReadAll(resp.Body)
if err != nil {
return "", nil, fmt.Errorf("read response failed: %w", err)
}
var result struct {
ID string `json:"id"`
TaskID string `json:"task_id"`
@@ -2679,10 +2733,12 @@ func prepareFirstFrame(ctx context.Context, body map[string]any, segIdx int, gen
firstFrameItem[urlField] = resolvedPath
// 持久化到 DB（崩溃恢复用）
if updatedScript, err := json.Marshal(body); err == nil {
_ = dao.GenerationTask.UpdateFields(ctx, genTaskId, g.Map{
"script": string(updatedScript),
})
if updatedScript, err := json.Marshal(body); err != nil {
g.Log().Warningf(ctx, "marshal updated script failed: %v", err)
} else if ue := dao.GenerationTask.UpdateFields(ctx, genTaskId, g.Map{
"script": string(updatedScript),
}); ue != nil {
g.Log().Errorf(ctx, "persist updated script to task %d failed: %v", genTaskId, ue)
}
return
}
@@ -2802,6 +2858,12 @@ func rebuildTasksMedia(ctx context.Context, epId int64) {
g.Log().Warningf(ctx, "rebuildTasksMedia: list tasks failed: %v", err)
return
}
// 收集需要更新的任务，用单条 CASE 语句批量更新（避免循环单条 UPDATE）
var (
ids []int64
cases []string
args []any
)
for _, t := range tasks {
if t.Status != consts.TaskStatusPending || t.Script == "" {
continue
@@ -2810,11 +2872,22 @@ func rebuildTasksMedia(ctx context.Context, epId int64) {
if !modified {
continue
}
if _, e := g.DB().Model(consts.TableNameGenerationTask).Ctx(ctx).
Data(g.Map{"script": string(newScript)}).
Where("id", t.Id).Update(); e != nil {
g.Log().Warningf(ctx, "rebuildTasksMedia: update task %d script failed: %v", t.Id, e)
}
ids = append(ids, t.Id)
cases = append(cases, "WHEN id=? THEN ?")
args = append(args, t.Id, string(newScript))
}
if len(ids) == 0 {
return
}
for _, id := range ids {
args = append(args, id)
}
sql := "UPDATE " + consts.TableNameGenerationTask +
" SET script = CASE id " + strings.Join(cases, " ") +
" END, updated_at = datetime('now','localtime') WHERE id IN (" +
strings.TrimSuffix(strings.Repeat("?,", len(ids)), ",") + ")"
if _, e := g.DB().Exec(ctx, sql, args...); e != nil {
g.Log().Warningf(ctx, "rebuildTasksMedia: batch update %d tasks failed: %v", len(ids), e)
}
}
shortdrama/service/model_config_service.go
+15 -4
View File
@@ -23,7 +23,11 @@ var ModelConfigService = new(modelConfigService)
func (s *modelConfigService) getModelList(ctx context.Context) []*entity.ModelConfig {
list, err := dao.ModelConfig.GetAll(ctx)
if err != nil || len(list) == 0 {
if err != nil {
g.Log().Warningf(ctx, "load model config list failed: %v", err)
return make([]*entity.ModelConfig, 0)
}
if len(list) == 0 {
return make([]*entity.ModelConfig, 0)
}
return list
@@ -55,15 +59,22 @@ func (s *modelConfigService) GetModelListResponse(ctx context.Context) *dto.GetM
func (s *modelConfigService) SaveModelConfig(ctx context.Context, req *dto.SaveModelConfigReq) error {
cfg := new(entity.ModelConfig)
if req.Id > 0 {
existing, _ := dao.ModelConfig.GetAll(ctx)
existing, err := dao.ModelConfig.GetAll(ctx)
if err != nil {
return err
}
for _, m := range existing {
if m.Id == req.Id {
gconv.Struct(m, cfg)
if err := gconv.Struct(m, cfg); err != nil {
return err
}
break
}
}
}
gconv.Struct(req, cfg)
if err := gconv.Struct(req, cfg); err != nil {
return err
}
if req.Schema != nil {
cfg.Schema = req.Schema.String()
shortdrama/service/payment_order_service.go
+45 -29
View File
@@ -15,6 +15,7 @@ import (
"sort"
"strings"
"time"
"video-factory/shortdrama/consts"
"video-factory/shortdrama/dao"
"video-factory/shortdrama/model/entity"
@@ -113,18 +114,22 @@ func (s *paymentService) Prepay(ctx context.Context, userId int64, amount int64,
}
if err != nil {
_ = dao.PaymentOrder.UpdateFail(ctx, id, fmt.Sprintf(`{"error":"%s"}`, err.Error()))
if ufErr := dao.PaymentOrder.UpdateFail(ctx, id, fmt.Sprintf(`{"error":"%s"}`, err.Error())); ufErr != nil {
g.Log().Errorf(ctx, "update order %s to failed failed: %v", orderNo, ufErr)
}
return nil, "", "", "", err
}
if channel != "offline" {
_, _ = dao.PaymentChannelTrade.Insert(ctx, &entity.PaymentChannelTrade{
if _, trErr := dao.PaymentChannelTrade.Insert(ctx, &entity.PaymentChannelTrade{
OrderId: id,
Channel: channel,
CodeUrl: codeUrl,
PrepayId: prepayJson,
ChannelResponse: "{}",
})
}); trErr != nil {
g.Log().Warningf(ctx, "insert channel trade for order %s failed: %v", orderNo, trErr)
}
}
return order, codeUrl, prepayJson, redirectUrl, nil
@@ -166,13 +171,17 @@ func (s *paymentService) CreateRenewalOrder(ctx context.Context, agentId int64,
NotifyRaw: fmt.Sprintf(`{"duration":%d}`, duration),
}
err = g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
err = g.DB(consts.DbGroupFinance).Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
ap, _ := dao.AgentProfile.Get(ctx, agentId)
if ap != nil {
ap.RegionProtected = pricing.Protected == 1
ap.MaxCustomers = pricing.MaxCustomers
_ = dao.AgentProfile.Update(ctx, ap)
_ = dao.User.UpdateFields(ctx, agentId, g.Map{"region": pricing.Region})
if e := dao.AgentProfile.Update(ctx, ap); e != nil {
return e
}
if e := dao.User.UpdateFields(ctx, agentId, g.Map{"region": pricing.Region}); e != nil {
return e
}
}
id, e := dao.PaymentOrder.Insert(ctx, order)
if e != nil {
@@ -200,7 +209,7 @@ func (s *paymentService) ConfirmOffline(ctx context.Context, orderNo string) err
return errors.New("only offline payment orders can be confirmed")
}
return g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
return g.DB(consts.DbGroupFinance).Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
if err := dao.PaymentOrder.UpdateSuccess(ctx, order.Id, `{"confirm":"manual"}`); err != nil {
return err
}
@@ -250,26 +259,27 @@ func (s *paymentService) HandleNotify(ctx context.Context, channel string, body
}
rawStr := string(body)
if err := dao.PaymentOrder.UpdateSuccess(ctx, order.Id, rawStr); err != nil {
return err
}
if _, err := dao.PaymentChannelTrade.Insert(ctx, &entity.PaymentChannelTrade{
OrderId: order.Id,
Channel: channel,
TradeNo: tradeNo,
ChannelResponse: rawStr,
}); err != nil {
return err
}
switch order.OrderType {
case "recharge":
return TransactionService.Recharge(ctx, order.UserId, order.Amount, orderNo, fmt.Sprintf("支付充值 %.2f 元", float64(order.Amount)/100))
case "renewal":
return AgentService.RenewAgentByOrder(ctx, order.UserId, order.NotifyRaw)
}
return nil
// 订单状态更新、渠道流水写入、余额/续费变更放在同一事务内，保证原子性
return g.DB(consts.DbGroupFinance).Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
if err := dao.PaymentOrder.UpdateSuccess(ctx, order.Id, rawStr); err != nil {
return err
}
if _, err := dao.PaymentChannelTrade.Insert(ctx, &entity.PaymentChannelTrade{
OrderId: order.Id,
Channel: channel,
TradeNo: tradeNo,
ChannelResponse: rawStr,
}); err != nil {
return err
}
switch order.OrderType {
case "recharge":
return TransactionService.Recharge(ctx, order.UserId, order.Amount, orderNo, fmt.Sprintf("支付充值 %.2f 元", float64(order.Amount)/100))
case "renewal":
return AgentService.RenewAgentByOrder(ctx, order.UserId, order.NotifyRaw)
}
return nil
})
}
// GetStatus 查询订单状态
@@ -318,7 +328,10 @@ func (s *paymentService) callWechat(ctx context.Context, order *entity.PaymentOr
return "", "", fmt.Errorf("wechat order failed: %w", err)
}
defer resp.Body.Close()
b, _ := io.ReadAll(resp.Body)
b, err := io.ReadAll(resp.Body)
if err != nil {
return "", "", fmt.Errorf("wechat order response read failed: %w", err)
}
result := s.xmlToMap(string(b))
if result["return_code"] != "SUCCESS" || result["result_code"] != "SUCCESS" {
@@ -391,7 +404,10 @@ func (s *paymentService) callAlipay(ctx context.Context, order *entity.PaymentOr
return "", "", fmt.Errorf("alipay order failed: %w", err)
}
defer resp.Body.Close()
b, _ := io.ReadAll(resp.Body)
b, err := io.ReadAll(resp.Body)
if err != nil {
return "", "", fmt.Errorf("alipay order response read failed: %w", err)
}
var result struct {
Response struct {
shortdrama/service/transaction_service.go
+29 -21
View File
@@ -3,9 +3,13 @@ package service
import (
"context"
"fmt"
"video-factory/shortdrama/consts"
"video-factory/shortdrama/dao"
"video-factory/shortdrama/model/entity"
"github.com/gogf/gf/v2/database/gdb"
"github.com/gogf/gf/v2/frame/g"
)
type transactionService struct{}
@@ -20,31 +24,35 @@ func (s *transactionService) Insert(ctx context.Context, data *entity.AccountTra
return dao.AccountTransaction.Insert(ctx, data)
}
// Recharge 充值：余额更新与流水写入放在同一事务内，保证原子性。
// 若调用方已开启 finance 事务（如确认支付），会自动复用外层事务。
func (s *transactionService) Recharge(ctx context.Context, customerId int64, amount int64, orderNo, remark string) error {
if amount <= 0 {
return fmt.Errorf("recharge amount must be greater than 0")
}
cp, err := dao.CustomerProfile.Get(ctx, customerId)
if err != nil || cp == nil {
return fmt.Errorf("customer not found")
}
newBalance := cp.Balance + amount
if err := dao.CustomerProfile.UpdateBalance(ctx, customerId, newBalance); err != nil {
return g.DB(consts.DbGroupFinance).Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
cp, err := dao.CustomerProfile.Get(ctx, customerId)
if err != nil || cp == nil {
return fmt.Errorf("customer not found")
}
newBalance := cp.Balance + amount
if err := dao.CustomerProfile.UpdateBalance(ctx, customerId, newBalance); err != nil {
return err
}
rk := remark
if rk == "" {
rk = fmt.Sprintf("充值 %.2f 元", float64(amount)/100)
}
_, err = dao.AccountTransaction.Insert(ctx, &entity.AccountTransaction{
UserId: customerId,
Type: "recharge",
Amount: amount,
BalanceBefore: cp.Balance,
BalanceAfter: newBalance,
OrderNo: orderNo,
Remark: rk,
CreatedBy: "system",
})
return err
}
rk := remark
if rk == "" {
rk = fmt.Sprintf("充值 %.2f 元", float64(amount)/100)
}
_, err = dao.AccountTransaction.Insert(ctx, &entity.AccountTransaction{
UserId: customerId,
Type: "recharge",
Amount: amount,
BalanceBefore: cp.Balance,
BalanceAfter: newBalance,
OrderNo: orderNo,
Remark: rk,
CreatedBy: "system",
})
return err
}
shortdrama/service/user_model_config_service.go
+18 -4
View File
@@ -54,7 +54,10 @@ func (s *userModelConfigService) GetUserConfig(ctx context.Context, userId int64
}
func (s *userModelConfigService) SaveUserConfigs(ctx context.Context, userId int64, items []*dto.SaveUserModelConfigItem) error {
existingList, _ := dao.UserModelConfig.GetByUserId(ctx, userId)
existingList, err := dao.UserModelConfig.GetByUserId(ctx, userId)
if err != nil {
return err
}
existingByType := make(map[string]*entity.UserModelConfig)
for _, ec := range existingList {
if ec.ModelType != "" {
@@ -112,7 +115,10 @@ func (s *userModelConfigService) SaveUserConfig(ctx context.Context, userId int6
}
}
existing, _ := dao.UserModelConfig.GetByModelConfigId(ctx, userId, cfg.ModelConfigId)
existing, err := dao.UserModelConfig.GetByModelConfigId(ctx, userId, cfg.ModelConfigId)
if err != nil {
return err
}
if existing != nil {
cfg.Id = existing.Id
cfg.CreatedAt = existing.CreatedAt
@@ -190,7 +196,11 @@ func (s *userModelConfigService) GetMergedConfig(ctx context.Context, userId int
}
func (s *userModelConfigService) GetUserModelList(ctx context.Context, userId int64) *dto.GetUserModelListRes {
userCfgs, _ := dao.UserModelConfig.GetByUserId(ctx, userId)
userCfgs, err := dao.UserModelConfig.GetByUserId(ctx, userId)
if err != nil {
g.Log().Warningf(ctx, "list user model configs of user %d failed: %v", userId, err)
return &dto.GetUserModelListRes{List: make([]*dto.UserModelItem, 0)}
}
if len(userCfgs) == 0 {
return &dto.GetUserModelListRes{List: make([]*dto.UserModelItem, 0)}
}
@@ -232,7 +242,11 @@ func (s *userModelConfigService) GetUserModelList(ctx context.Context, userId in
func (s *userModelConfigService) GetUserModelListPage(ctx context.Context, userId int64, page, pageSize int, modelType, keyword string) *dto.GetUserModelListRes {
// 先查用户配置，获取 model_config_id 列表
userCfgs, _ := dao.UserModelConfig.GetByUserId(ctx, userId)
userCfgs, err := dao.UserModelConfig.GetByUserId(ctx, userId)
if err != nil {
g.Log().Warningf(ctx, "list user model configs of user %d failed: %v", userId, err)
return &dto.GetUserModelListRes{List: make([]*dto.UserModelItem, 0)}
}
if len(userCfgs) == 0 {
return &dto.GetUserModelListRes{List: make([]*dto.UserModelItem, 0)}
}
shortdrama/service/user_service.go
+5 -1
View File
@@ -9,6 +9,7 @@ import (
"video-factory/shortdrama/dao"
"video-factory/shortdrama/model/entity"
"github.com/gogf/gf/v2/frame/g"
"github.com/golang-jwt/jwt/v5"
"golang.org/x/crypto/bcrypt"
)
@@ -34,7 +35,10 @@ func (s *userService) Login(ctx context.Context, account, password string) (*ent
var agentId int64
if user.Role == "customer" {
cp, _ := dao.CustomerProfile.Get(ctx, user.Id)
cp, err := dao.CustomerProfile.Get(ctx, user.Id)
if err != nil {
g.Log().Warningf(ctx, "login: get customer profile %d failed: %v", user.Id, err)
}
if cp != nil {
agentId = cp.AgentId
}