LongArticlesJob/long_articles_job.py

from argparse import ArgumentParser

from tasks.crawler_tasks.crawler_video.crawler_piaoquan_videos import (
    CrawlerPiaoQuanVideos,
)
from tasks.crawler_tasks.crawler_video.crawler_sohu_videos import CrawlerSohuHotVideos
from tasks.crawler_tasks.crawler_video.crawler_sohu_videos import (
    CrawlerSohuRecommendVideos,
)
from tasks.crawler_tasks.crawler_video.crawler_sph_videos import (
    CrawlerChannelAccountVideos,
)
from tasks.crawler_tasks.crawler_video.crawler_gzh_videos import CrawlerGzhMetaVideos
from tasks.data_tasks.fwh_data_recycle import FwhGroupPublishRecordManager
from tasks.data_tasks.fwh_data_recycle import SaveFwhDataToDatabase
from tasks.data_tasks.fwh_data_recycle import FwhGroupPublishMonitor
from tasks.monitor_tasks.kimi_balance_monitor import check_kimi_balance
from tasks.publish_tasks.top_article_generalize import (
    TopArticleGeneralizeFromArticlePool,
)


def run_piaoquan_video_crawler():
    crawler = CrawlerPiaoQuanVideos()
    crawler.deal()


def run_sohu_video_crawler():
    # step1, crawl sohu hot videos
    crawler_sohu_hot_videos = CrawlerSohuHotVideos()
    crawler_sohu_hot_videos.deal()

    # step2, crawl sohu recommend videos
    crawler_sohu_recommend_videos = CrawlerSohuRecommendVideos()
    crawler_sohu_recommend_videos.deal()


def run_sph_video_crawler():
    crawler_channel_account_videos = CrawlerChannelAccountVideos()
    crawler_channel_account_videos.deal()


def run_fwh_data_manager():
    # 1. 从 aigc 获取数据
    fwh_group_publish_record_manager = FwhGroupPublishRecordManager()
    fwh_group_publish_record_manager.deal()

    # 2. 监测报警
    fwh_group_publish_monitor = FwhGroupPublishMonitor()
    fwh_group_publish_monitor.deal()

    # 3. 保存数据到数据库
    save_fwh_data_to_database = SaveFwhDataToDatabase()
    save_fwh_data_to_database.deal()


def run_top_article_generalize_from_article_pool():
    task = TopArticleGeneralizeFromArticlePool()
    task.deal()


def crawler_gzh_meta_videos():
    task = CrawlerGzhMetaVideos()
    task.deal()


def main():
    """
    run long_articles_job
    """
    parser = ArgumentParser()
    parser.add_argument("--task_name", help="which task you want to run")
    parser.add_argument("--run_date", help="task specify run date")

    args = parser.parse_args()

    task_name = args.task_name
    if task_name is None:
        print("task_name cannot be None")
        return
    else:
        match task_name:
            case "run_piaoquan_video_crawler":
                run_piaoquan_video_crawler()
            case "run_sohu_video_crawler":
                run_sohu_video_crawler()
            case "run_check_kimi_balance":
                check_kimi_balance()
            case "run_fwh_data_manager":
                run_fwh_data_manager()
            case "run_sph_video_crawler":
                run_sph_video_crawler()
            case "top_article_generalize":
                run_top_article_generalize_from_article_pool()
            case "crawler_gzh_meta_videos":
                crawler_gzh_meta_videos()
            case _:
                print("task_name cannot be None")


if __name__ == "__main__":
    main()