PostHog Weekly Digest 架构解析Temporal 工作流 Redis 中间层驱动的个性化周报引擎【免费下载链接】posthog:hedgehog: PostHog is the leading platform for building self-driving products. Our developer tools – AI observability, analytics, session replay, flags, experiments, error tracking, logs, and more – capture all the context agents need to diagnose problems, uncover opportunities, and ship fixes. Steer it all from Slack, web, desktop, or the MCP.项目地址: https://gitcode.com/GitHub_Trending/po/posthog每周自动生成并发送一封过去一周项目动态总结邮件是 PostHog 为所有客户提供的默认产品能力之一。本文以仓库中 posthog/temporal/weekly_digest/README.md 为主干结合该目录下的完整源码深入讲解这套周报系统的整体架构、Redis 存储结构、三类数据团队级 / 组织级 / 用户级的组织方式以及如何向周报中新增团队级字段与用户级字段。读完本文你将掌握一个大规模多租户邮件系统的经典实现范式用 Temporal 子工作流做两阶段流水线先生成、后发送用 Redis 作为批次间的数据交换层用 Pydantic 模型做类型化的数据契约。一、系统定位与总体架构Weekly Digest 是 PostHog 面向所有客户发送的一封周期性邮件汇总其 PostHog 项目中过去一周发生的关键活动例如新建的仪表盘、新定义的事件、启动/完成的实验、新增的特性开关Feature Flag、有趣的 Session Replay 筛选器、即将过期的录制、启动的调研问卷、事件量与活跃用户的周环比WoW趋势、以及新增的 Error Tracking 问题等。整套流程由两个 Temporal 工作流协作完成定义见 workflows.pyGenerateDigestDataWorkflowTemporal 名称generate-digest-data负责生成所有摘要数据并写入 RedisSendWeeklyDigestWorkflowTemporal 名称send-weekly-digest从 Redis 读取数据为每个用户渲染并发送个性化邮件。两者之上还有一个编排入口WeeklyDigestWorkflowTemporal 名称weekly-digest它负责计算摘要周期上周时间窗与digest_key、解析输入参数然后依次以子工作流方式调用上述两个工作流。三个工作流均继承自PostHogWorkflow基类posthog/temporal/common/base.py并被统一注册在init.py 的WORKFLOWS/ACTIVITIES列表中供start_temporal_worker与start_temporal_workflow等管理命令加载参见 start_temporal_workflow.py 中from posthog.temporal.weekly_digest import WORKFLOWS as WEEKLY_DIGEST_WORKFLOWS。从源码结构看这套weekly_digest模块同时是其他产品线周报的复用基座例如 products/web_analytics/backend/temporal/weekly_digest/ 实现了 Web Analytics 的周报工作流wa_weekly_digest其管理命令测试甚至专门断言了WA digests 已随 weekly digest 一起注册test_start_temporal_worker.py。为什么需要两阶段 Redis 中间层周报要处理的是全量组织 → 全量团队 → 全量用户的扇出fan-out计算。若在发送阶段实时聚合每个用户的数据会产生海量重复查询而把团队级数据的计算与用户级邮件的发送解耦成两个可独立重试、独立扩容的工作流中间用 Redis 落盘中间结果则可以团队级数据只计算一次所有用户共享同一份OrganizationDigest生成阶段可以按团队 id 区间并行切分发送阶段可以按组织区间并行切分生成失败时只需重跑生成工作流发送阶段完全幂等。二、Redis 存储结构三类 Key 与类型安全约定所有 Redis Key 都通过 keys.py 中的辅助函数生成严禁在代码中硬编码 Key 字符串必须使用类型化枚举与生成函数。Key 统一以{digest_key}为前缀例如weekly-digest-2024-01由 ISO 年-周号生成见workflows.py中fweekly-digest-{year}-{week}。2.1 团队级数据Team-level通过team_data_key(digest_key, TeamDataKey.*, team_id)生成映射关系与枚举定义完全一致对比 keys.pyTeamDataKey枚举Key 模式内容DASHBOARDS{digest_key}-dashboards-{team_id}新建的仪表盘EVENT_DEFINITIONS{digest_key}-event-definitions-{team_id}新定义的事件EXPERIMENTS_LAUNCHED{digest_key}-experiments-launched-{team_id}启动的实验EXPERIMENTS_COMPLETED{digest_key}-experiments-completed-{team_id}完成的实验EXTERNAL_DATA_SOURCES{digest_key}-external-data-sources-{team_id}新增的外部数据源FEATURE_FLAGS{digest_key}-feature-flags-{team_id}新增的特性开关SAVED_FILTERS{digest_key}-saved-filters-{team_id}有趣的 Replay 筛选器EXPIRING_RECORDINGS{digest_key}-expiring-recordings-{team_id}即将过期录制的数量SURVEYS_LAUNCHED{digest_key}-surveys-launched-{team_id}启动的调研问卷USAGE_TRENDS{digest_key}-usage-trends-{team_id}事件量 活跃用户周环比ERROR_ISSUES{digest_key}-error-issues-{team_id}新增的 Error Tracking 问题其中USAGE_TRENDS与ERROR_ISSUES是注释中标注的新信号new signalsTeamDigest为它们提供了默认值UsageTrends()与ErrorIssueList(root[])保证旧调用方与旧测试仍能正常构造对象见 types.py。2.2 组织级数据Organization-level通过org_digest_key(digest_key, org_id)生成Key 模式内容{digest_key}-{org_id}一个OrganizationDigest内含该组织下所有团队的摘要OrganizationDigest在生成阶段由generate_organization_digest_batch活动一次性聚合写盘活动名generate-organization-digest-batch发送阶段每个用户都从这份共享数据出发做个性化裁剪。2.3 用户级数据User-level通过user_data_key(digest_key, UserDataKey.*, user_id)生成见 keys.pyUserDataKey枚举Key 模式内容NOTIFY_TEAMS{digest_key}-user-notify-{user_id}Redis SET记录该用户需要接收通知的团队 id 集合PRODUCT_SUGGESTION{digest_key}-product-suggestion-{user_id}单个DigestProductSuggestion用于产品推荐注意NOTIFY_TEAMS使用r.sadd写入一个用户可属于多个团队而其余数据使用r.setex写入 JSON 字符串两者都设置了CommonInput.redis_ttl的 TTL——默认 3 天3600 * 24 * 3见 types.py。2.4 公共参数CommonInput所有活动输入都携带一个CommonInputtypes.py字段默认值说明redis_ttl3600 * 24 * 33 天Redis Key 的过期时间redis_hostNone运行时回退到环境变量WEEKLY_DIGEST_REDIS_HOST默认localhost周报专用 Redis 主机redis_portNone运行时回退到WEEKLY_DIGEST_REDIS_PORT默认6379周报专用 Redis 端口batch_size2500团队/组织分批切分的批大小django_redis_urlNone运行时回退到settings.REDIS_URLDjango 缓存 Redis 地址仅 Replay 筛选器计数需要环境变量回退逻辑位于WeeklyDigestWorkflow.runworkflows.pyWEEKLY_DIGEST_REDIS_HOST、WEEKLY_DIGEST_REDIS_PORT用于指定周报专用 Redis避免与业务缓存互相污染django_redis_url则用于读取 Django 缓存中已算好的播放列表计数。三、数据流从团队切分到邮件发送README 给出的完整数据流如下此图与 workflows.py 中的实际编排一一对应┌─────────────────────────────────────────────────────────────────────────────┐ │ GenerateDigestDataWorkflow │ ├─────────────────────────────────────────────────────────────────────────────┤ │ │ │ 1. Cut teams into id-range batches, count orgs for batching │ │ │ │ 2. Generate team-level data (parallel per batch): │ │ ├── generate_dashboard_lookup │ │ ├── generate_event_definition_lookup │ │ ├── generate_experiment_launched_lookup │ │ ├── generate_experiment_completed_lookup │ │ ├── generate_external_data_source_lookup │ │ ├── generate_feature_flag_lookup │ │ ├── generate_survey_lookup │ │ ├── generate_filter_lookup │ │ ├── generate_recording_lookup │ │ ├── generate_user_notification_lookup │ │ └── generate_product_suggestion_lookup │ │ │ │ 3. Aggregate into org digests: │ │ └── generate_organization_digest_batch │ │ │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ Redis Storage │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ SendWeeklyDigestWorkflow │ ├─────────────────────────────────────────────────────────────────────────────┤ │ │ │ For each organization (batched): │ │ 1. Load OrganizationDigest from Redis │ │ 2. For each org member: │ │ a. Load users notification team set │ │ b. Load users product suggestion │ │ c. Create UserSpecificDigest via org_digest.for_user() │ │ d. Render payload and send via PostHog capture event │ │ │ └─────────────────────────────────────────────────────────────────────────────┘3.1 生成阶段的关键设计团队 id 区间切分list_team_id_ranges活动activities.py一次性对query_team_ids_for_digest()做主键索引扫描按batch_size切成若干[start, end)半开区间_cut_team_id_ranges。这样每个 generator 活动都通过主键谓词filter(id__gtestart, id__ltend)直接定位到批首行避免 LIMIT/OFFSET 每批都要构建并丢弃批前所有行的性能问题activities.py。并行扇出GenerateDigestDataWorkflow.run用itertools.product(team_id_ranges, generators)生成区间 × 生成器的全笛卡尔组合通过asyncio.gather并发调度所有活动每个活动配置了start_to_close_timeout1 小时、heartbeat_timeout2 分钟、RetryPolicy(maximum_attempts2, initial_interval1 分钟)workflows.py。活动内部全部使用Heartbeater()保持心跳配合日志上下文bind_contextvars记录digest_key、period_start、period_end与团队区间。通用的 lookup 模板除 Replay 相关与 usage trends 外绝大多数团队级数据都走同一个模板函数generate_digest_data_lookupactivities.py它接收三个参数key_kindTeamDataKey枚举、query_func从 queries.py 传入的查询函数、resource_type对应的 Pydantic RootModel 列表类型。流程为查询该团队在周期内新增的数据 → 封装成列表模型 →r.setex写入team_data_key(...)并设置 TTL单团队失败只记录 warning 并跳过不影响批次。3.2 特殊数据源的生成逻辑三个团队级字段没有走通用模板值得单独说明Replay 筛选器generate-filter-lookup除周报 Redis 外还连接django_redis_url指定的 Django 缓存从PLAYLIST_COUNT_REDIS_PREFIX前缀下批量mget播放列表的录制计数PlaylistCount把recording_count与more_available回填到筛选器上最后按录制数降序排列FilterList.order_by_recording_count。解析失败视同无计数activities.py。即将过期的录制generate-recording-lookup使用 ClickHouse 客户端执行SessionReplayEvents.count_soon_to_expire_sessions_query参数含ttl_threshold10天响应经ClickHouseResponse模型解析后取出RecordingCountactivities.py。用量趋势generate-usage-trends-lookup在离线集群Workload.OFFLINE上执行一段内嵌的 HogQL 查询USAGE_TRENDS_QUERY用countIf/uniqExactIf(person_id, ...)在一次扫描中同时统计本周 vs 上周的事件量与去重活跃用户数filterTestAccountsTrue继承团队自己的测试账号过滤规则使数字与团队在 Product Analytics 中看到的一致。当前周无事件events_current 0的团队直接跳过不写。周环比变化率通过compute_week_over_week_change来自 posthog/tasks/email_utils.py计算并归一为directionup/down/flat与change_pcthas_baselineFalse表示上周无数据可比邮件模板会渲染new而非把 0→N 增长混同于无变化activities.py。此外generate_usage_trends_lookup有一个值得借鉴的健壮性处理若某批次attempted 0且全部失败例如离线集群故障或查询语句被写坏它会主动raise RuntimeError让活动重试并最终让运行显式失败——否则全部失败会与没有活跃团队表象相同从而静默发出空的 usage 区块activities.py。3.3 组织级聚合所有团队级数据生成完毕后count_organizations统计组织总数按batch_size切成(start, end)批次再由asyncio.gather并发执行generate_organization_digest_batch。该活动对每个组织下的每个团队用一次r.mget批量读取全部 11 个团队级 Key与defaults列表空列表默认值逐位 zip某 Key 缺失result is None时用对应类型的空默认值兜底。随后组装出TeamDigest列表并写入org_digest_key(...)activities.py。3.4 发送阶段SendWeeklyDigestWorkflow先count_organizations再按组织区间切批并发执行send_weekly_digest_batchworkflows.py。send_weekly_digest_batchactivities.py内部的关键逻辑从 Redis 读取OrganizationDigest缺失则跳过内容门槛org_digest.is_empty() or org_digest.count_items() DIGEST_ITEM_COUNT_THRESHOLD阈值为 4时跳过。注意TeamDigest._fields()有意不统计usage_trends——它是几乎每个活跃团队都有的环境性上下文计入会把几乎所有组织都推过阈值从而人人收到邮件而error_issues被计入因为出现新的生产错误本身就值得一封邮件types.py幂等防重发通过MessagingRecord.objects.aget_or_create(raw_emailforg_{organization.id}, campaign_keyinput.digest.key)记录发送状态若sent_at已存在且未设置allow_already_sent则跳过成功后以 100 条一批abulk_update回写sent_atRECORD_BATCH_SIZE 100遍历组织成员读取用户的NOTIFY_TEAMS集合r.smembers与PRODUCT_SUGGESTION构建UserDigestContext调用org_digest.for_user(user_notify_teams, user_context)得到个性化UserSpecificDigest同样应用空或少于 4 项则跳过的用户级门槛渲染 payload 后实际发送并非直接调用 SMTP而是通过 PostHog 自身的 capture 事件ph_client.capture(distinct_iduser.distinct_id, eventtransactional email, propertiespayload, groups{...})——注释说明只有 US 部署会把邮件事件转发给 customer.ioactivities.py。dry_run模式下只打印日志不发送。四、关键类型Key Types4.1Digest与WeeklyDigestInputDigesttypes.py描述一次摘要周期含key如weekly-digest-2026-37、period_start、period_end并提供render_payload()把周期信息序列化进邮件 payload。WeeklyDigestInputtypes.py是入口工作流的 CLI 输入支持四个开关字段默认值说明dry_runFalse试运行只打日志不真正发送skip_generateFalse跳过生成阶段直接复用 Redis 中已生成的数据digest_key_overrideNone覆盖自动生成的weekly-digest-{年}-{周}Key 前缀allow_already_sentFalse允许对已发送过的组织再次发送绕过幂等检查WeeklyDigestWorkflow.parse_inputs从管理命令 CLI 接收 JSON 字符串数组并model_validate_json解析workflows.py这说明可通过start_temporal_workflow管理命令以 JSON 形式传入参数启动参见 start_temporal_workflow.py 的workflowinputs参数设计。4.2OrganizationDigest/UserDigestContext/UserSpecificDigestOrganizationDigesttypes.py组织级基础摘要含id、name、created_at与team_digests列表整体序列化后存 Redis。其for_user(user_teams, context)方法把team_digests过滤为team_digest.id in user_teams的子集并挂上用户上下文返回UserSpecificDigest。UserDigestContexttypes.py所有用户级数据的容器当前只有product_suggestion一个字段注释明确指示在这里添加新的用户级字段class UserDigestContext(BaseModel): product_suggestion: DigestProductSuggestion | None None # Add new user-specific fields hereUserSpecificDigest发送时刻动态计算、不落 Redis的个性化视图。其render_payload(digest)遍历非空团队摘要仅当产品建议与团队匹配product_suggestion.team_id td.id时才把建议挂到该团队的 payload 上最终输出含organization_name、organization_id、teams、scope: user、template_name: weekly_digest_report、period与digest_regionget_instance_region()的完整邮件载荷types.py。4.3TeamDigest与资源模型TeamDigesttypes.py聚合单个团队的全部摘要字段并提供三个重要方法_fields()返回参与内容计数的资源列表排除 usage_trends理由见 3.4is_empty()/count_items()判断摘要是否为空及内容条数render_payload(product_suggestion)把各类列表model_dump()后组织成report字典new_dashboards、new_event_definitions、new_experiments_launched等可选用产品建议追加new_product_suggestion。各类资源都是轻量 Pydantic 模型DigestDashboardname/id、DigestEventDefinitionname/UUID id、DigestExperimentname/id/start_date/end_date、DigestExternalDataSourcesource_type/UUID id、DigestFeatureFlagname/id/key、DigestFiltername/short_id/view_count/recording_count/more_available、DigestSurveyname/UUID id/description/start_date、DigestErrorIssuename/UUID id名称为空时回退为 Untitled issue、以及UsageTrendMetric/UsageTrends。列表形态统一用RootModel包裹如DashboardList、FilterList并汇总为DigestResourceTypeTypeAliastypes.py。4.4DigestProductSuggestion产品推荐模型types.py包含team_id建议针对的项目、product_path产品名如 Session replay、reason_text人类可读的推荐理由。当 campaign 没有手写推广文案时回退到与导航卡片一致的默认文案DEFAULT_PRODUCT_SUGGESTION_TEXT。其生成逻辑在generate_product_suggestion_lookupactivities.py按组织缓存ProductPushCampaign仅statusACTIVE且started_at period_end的在跑 campaign见 queries.py通过resolve_product_path/project_uses_productproducts/growth/backend/product_push/selection.py判断团队是否已使用该产品已使用则不推送且尊重user.allow_sidebar_suggestions is False的退出开关每个用户只存第一个建议。五、如何新增一个团队级字段扩展指南团队级字段属于项目/团队维度的数据仪表盘、特性开关等。每种字段类型应有独立的 activity 方法以便关注点分离并支持并行执行。以新增ALERTS告警为例按 README 与源码的 7 个步骤1. 在keys.py中为TeamDataKey增加枚举值class TeamDataKey(StrEnum): # ... existing keys ... ALERTS alerts # new2. 在types.py中创建新数据类型的 Pydantic 模型例如DigestAlert、AlertList其中列表类需继承RootModel以匹配DigestResourceType。3. 在queries.py中新增查询函数从数据库按周期窗口取数。参考既有实现queries.py的约定时间窗口统一为created_at__gtperiod_start, created_at__lteperiod_end或实验类的start_date返回带team_id的.values(...)def query_new_alerts(period_start: datetime, period_end: datetime) - QuerySet: return Alert.objects.filter( created_at__gtperiod_start, created_at__lteperiod_end ).values(team_id, name, id)4. 在activities.py中创建新 activity复用通用模板generate_digest_data_lookupactivity.defn(namegenerate-alert-lookup) async def generate_alert_lookup(input: GenerateDigestDataBatchInput) - None: return await generate_digest_data_lookup( input, key_kindTeamDataKey.ALERTS, query_funcquery_new_alerts, resource_typeAlertList, )若需要每个团队最多 N 条可传per_team_limitN参考generate_error_issue_lookup中NEW_ERROR_ISSUES_PER_TEAM_LIMIT 5的用法见 activities.py。5. 在workflows.py中注册把generate_alert_lookup加入GenerateDigestDataWorkflow.run的generators列表workflows.py并同步在init.py 的ACTIVITIES中导出。generator 列表顺序无关紧要因为asyncio.gather会并发执行所有区间 × 生成器组合。6. 把字段加到TeamDigesttypes.py并更新generate_organization_digest_batch的聚合逻辑在r.mget的 Key 列表、defaults列表与TeamDigest(...)构造参数三处按TeamDataKey枚举顺序对齐追加——README 特别强调all_team_data_keys中的 Key 顺序与TeamDataKey枚举顺序一致这是最容易出错的一步activities.py。7. 更新TeamDigest.render_payload()在report字典中输出新字段邮件模板即可消费types.py。若新字段属于值得单独触发邮件的内容记得加入_fields()计数列表若属于环境性上下文则保持排除。六、如何新增一个用户级字段扩展指南用户级字段是按用户个性化的数据产品建议、通知偏好等。同样每种字段应有独立 activity 在生成阶段加载数据。以新增RECOMMENDATIONS为例1. 在keys.py中为UserDataKey增加枚举值class UserDataKey(StrEnum): # ... existing keys ... RECOMMENDATIONS recommendations # new2. 在types.py中创建 Pydantic 模型如UserRecommendations。3. 在queries.py中新增查询函数。4. 在activities.py中创建 activity遍历团队/用户查询数据并写入 Redisactivity.defn(namegenerate-user-recommendations-lookup) async def generate_user_recommendations_lookup(input: GenerateDigestDataBatchInput) - None: # Iterate through teams/users, query data, store in Redis using: key user_data_key(input.digest.key, UserDataKey.RECOMMENDATIONS, user.id)参考generate_user_notification_lookupactivities.py的写法它遍历区间内每个团队的所有有访问权限用户team.all_users_with_access()通过should_send_notification(user, NotificationSetting.WEEKLY_PROJECT_DIGEST.value, team.id)判断通知开关实现在 posthog/tasks/email.py先检查全局all_weekly_digest_disabled再按团队检查project_weekly_digest_disabled命中的用户用r.sadd把团队 id 加入其 NOTIFY_TEAMS 集合并设置 TTL。5. 在workflows.py中注册加入generators列表。6. 把字段加到UserDigestContexttypes.pyclass UserDigestContext(BaseModel): product_suggestion: DigestProductSuggestion | None None recommendations: UserRecommendations | None None # new field7. 在send_weekly_digest_batch中加载并挂到 context 上参考 activities.py 中 product_suggestion 的读取范式raw_recommendations await r.get( user_data_key(input.digest.key, UserDataKey.RECOMMENDATIONS, user.id) ) recommendations UserRecommendations.model_validate_json(raw_recommendations) if raw_recommendations else None user_context UserDigestContext( product_suggestionproduct_suggestion, recommendationsrecommendations, )8. 在UserSpecificDigest.render_payload()中使用把新字段渲染进邮件 payloadtypes.py。核心不变量for_user()的签名永远不变——所有用户级数据都经由UserDigestContext流入这保证了OrganizationDigest的存储格式稳定、发送阶段无需感知新增字段的细节。七、查询层与邮件触达的底层约定7.1 查询层的过滤规则queries.py 集中了所有数据访问几个值得注意的约定query_teams_for_digest()排除内部指标组织for_internal_metricsTrue与演示项目is_demoTrue只加载必要列并按id排序query_new_dashboards排除名称含 Generated Dashboard 的自动生成仪表盘query_new_feature_flags排除实验自动创建的特性开关名称含 Feature Flag for Experiment与调研目标 flagTargeting flag for surveyquery_saved_filters排除未命名/派生名为 (Untitled)/Unnamed 的播放列表、deletedTrue的、默认播放列表DEFAULT_PLAYLIST_NAMES与typecollection的并annotate出周期内的view_count实验的启动与完成是互斥的两个查询launched 排除同时期内也完成的实验completed 只看end_date落在窗口内的。7.2 发送渠道transactional email 事件值得强调的是这套周报的发送最终落到 PostHog 自身的 product analytics 事件上eventtransactional email按organization与instance分组而不是直接调用邮件服务。这意味着邮件内容、收件人、个性化数据都作为事件属性被 PostHog 记录后续由 US 部署转发到 customer.io 真正投递这也让整个流程天然可观测——每次发送都是一条可查询的事件。dry_run模式则让你在不上线真实邮件的情况下通过日志审查每个用户将要收到的完整 payload。八、小结这套架构的可复用要点从 posthog/temporal/weekly_digest 这个模块可以提炼出几个大规模周期性通知系统的通用设计两阶段流水线数据生成与消息发送解耦为两个 Temporal 工作流以 Redis 为中间层各自可独立重试、独立扩容、按区间并行切分类型化 Key 管理所有 Redis Key 由枚举 生成函数派生杜绝硬编码散落各处Pydantic 数据契约Redis 中流动的是可校验的 JSON 模型读取时model_validate_json缺省时有明确的空默认值兜底内容门槛与幂等DIGEST_ITEM_COUNT_THRESHOLD防止空周报打扰用户MessagingRecord.sent_at保证同一周期不会重复发送用户个性化与数据生成解耦OrganizationDigest存公共数据for_user()UserDigestContext在发送时按用户裁剪新增用户级字段不需要改动存储结构与for_user签名健壮性细节单团队失败跳过并告警、整批失败显式抛错usage trends、解析失败的缓存数据视同缺失playlist counts、活动全程心跳 重试策略。对于任何周期性地向海量多租户用户生成并投递个性化内容的系统周报、对账单、告警摘要等这套Temporal 编排 Redis 中转 Pydantic 契约 事件化投递的组合都是可以直接借鉴的成熟范本。【免费下载链接】posthog:hedgehog: PostHog is the leading platform for building self-driving products. Our developer tools – AI observability, analytics, session replay, flags, experiments, error tracking, logs, and more – capture all the context agents need to diagnose problems, uncover opportunities, and ship fixes. Steer it all from Slack, web, desktop, or the MCP.项目地址: https://gitcode.com/GitHub_Trending/po/posthog创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考