fix 邮件发送失败问题

.env

@@ -0,0 +1,14 @@
+ENV=dev
+
+DEBUG=true
+
+# 日志配置
+LOG_LEVEL=DEBUG
+LOG_TYPE=console
+
+# 数据库配置
+DB_HOST= 47.119.128.161                   # 192.168.1.200
+DB_PORT=19732
+DB_USER=postgres
+DB_PASS=postgres
+DB_NAME=peter

arlo.py

@@ -7,7 +7,6 @@ include exception tracebacks for easier debugging.
 
 import datetime
 import signal
-import sys
 import traceback
 from functools import partial
 from typing import Any
@@ -16,8 +15,8 @@ from apscheduler.events import EVENT_JOB_ERROR
 from apscheduler.schedulers.blocking import BlockingScheduler
 
 from config import config
-from log.log_manager import log, logger
 from task.manager_task import manager_task
+from utils import logger
 
 
 def _format_exception(exc: BaseException) -> str:
@@ -84,7 +83,7 @@ def main() -> None:
 
     # Graceful shutdown handlers
     def _shutdown(signum, frame):
-        log(f"Received signal {signum}. Shutting down scheduler...")
+        logger.info(f"Received signal {signum}. Shutting down scheduler...")
         try:
             scheduler.shutdown(wait=False)
         except Exception:
@@ -96,10 +95,10 @@ def main() -> None:
     signal.signal(signal.SIGTERM, _shutdown)
 
     try:
-        log("started successfully.")
+        logger.info("started successfully.")
         scheduler.start()  # 阻塞运行
     except (KeyboardInterrupt, SystemExit):
-        log("Shutting down ...")
+        logger.info("Shutting down ...")
         if scheduler.state == 1:
             try:
                 scheduler.shutdown(wait=False)

config/__init__.py

@@ -0,0 +1 @@
+from config.settings import settings

config/database.py

@@ -0,0 +1,18 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, scoped_session
+from config.settings import settings
+
+SQLALCHEMY_SYNC_URL = (
+    f"postgresql+psycopg://{settings.DB_USER}:{settings.DB_PASS}"
+    f"@{settings.DB_HOST}:{settings.DB_PORT}/{settings.DB_NAME}"
+)
+
+engine = create_engine(
+    SQLALCHEMY_SYNC_URL,
+    echo=False,       # 开发可改 True
+    future=True
+)
+
+SessionLocal = scoped_session(
+    sessionmaker(bind=engine, autoflush=False, autocommit=False)
+)

config/env_loader.py

@@ -0,0 +1,22 @@
+import os
+from dotenv import load_dotenv
+
+def load_env():
+    """
+    自动根据 ENV 加载对应的 .env 文件
+    """
+    base_file = ".env"
+    prod_file = ".env.prod"
+    test_file = ".env.test"
+
+    # 先加载基础 .env
+    if os.path.exists(base_file):
+        load_dotenv(base_file)
+
+    # 根据参数 ENV 再加载其他环境
+    env = os.getenv("ENV", "dev")
+
+    if env == "prod" and os.path.exists(prod_file):
+        load_dotenv(prod_file, override=True)
+    elif env == "test" and os.path.exists(test_file):
+        load_dotenv(test_file, override=True)

config/settings.py

@@ -0,0 +1,31 @@
+from pydantic_settings import BaseSettings
+from pydantic import Field
+from config.env_loader import load_env
+
+# 先加载 ENV & .env
+load_env()
+
+class Settings(BaseSettings):
+    # 环境
+    ENV: str = Field("dev")
+    DEBUG: bool = Field(True)
+
+    # 日志
+    LOG_LEVEL: str = Field("LOG_LEVEL")
+    LOG_FILE_PATH: str = Field("logs")
+    LOG_TYPE: str = Field("console")
+
+    # 数据库
+    DB_HOST: str
+    DB_PORT: int
+    DB_USER: str
+    DB_PASS: str
+    DB_NAME: str
+
+    class Config:
+        env_file = ".env"
+        env_file_encoding = "utf-8"
+
+
+# 全局唯一配置实例
+settings = Settings()

database/database.py

@@ -21,7 +21,14 @@ engine = create_engine(
         'keepalives_count': 5
     }
 )
-Base.metadata.create_all(engine)
+
+def get_engine():
+    return engine
+
+# 不在模块导入时自动创建表；提供显式创建函数
+def create_tables():
+    # 确保相关 model 模块已被导入并注册到 Base
+    Base.metadata.create_all(engine)
 
 @contextmanager
 def get_session():

database/tcontent/crud.py

@@ -0,0 +1,32 @@
+from database.tcontent.model import TContent
+
+def create_content(db, content: TContent):
+    db.add(content)
+    db.commit()
+    db.refresh(content)
+    return content
+
+def get_content_by_id(db, content_id: int):
+    return db.query(TContent).filter(TContent.id == content_id).first()
+
+def update_content(db, content_id: int, updates: dict):
+    content = db.query(TContent).filter(TContent.id == content_id).first()
+    if content:
+        for key, value in updates.items():
+            setattr(content, key, value)
+        db.commit()
+        db.refresh(content)
+    return content
+
+def delete_content(db, content_id: int):
+    content = db.query(TContent).filter(TContent.id == content_id).first()
+    if content:
+        db.delete(content)
+        db.commit()
+    return content
+
+def drop_table(db):
+    TContent.__table__.drop(db.get_bind(), checkfirst=True)
+
+def create_table(db):
+    TContent.__table__.create(db.get_bind(), checkfirst=True)

database/tcontent/debug.py

@@ -0,0 +1,15 @@
+import database.tcontent.model
+from database.database import get_session, create_tables
+from database.tcontent.crud import drop_table, get_content_by_id
+
+
+
+# create_tables()
+
+# with get_session() as db:
+#     task = get_content_by_id(db, 1)
+#     print(task)
+#     print(task.id)
+
+with get_session() as db:
+    drop_table(db)

database/tcontent/model.py

@@ -0,0 +1,12 @@
+from sqlalchemy import Column, Integer, String, Text, DateTime, func
+from database.database import Base
+
+class TContent(Base):
+    __tablename__ = 't_content'
+
+    id = Column(Integer, primary_key=True, autoincrement=True, comment='自动递增的唯一内容ID')
+    project = Column(String(64), nullable=False, comment='项目名称', index=True)
+    subject = Column(String(256), nullable=False, comment='主题')
+    content = Column(Text, nullable=True, comment='内容')
+    create_time = Column(DateTime, server_default=func.now(), nullable=False, comment='创建时间')
+    update_time = Column(DateTime, server_default=func.now(), onupdate=func.now(), nullable=False, comment='更新时间')

docker-compose.yml

@@ -5,3 +5,5 @@ services:
     image: arlo:latest
     container_name: arlo
     restart: always
+    environment:
+      - TZ=Asia/Shanghai  # 设置时区环境变量

mail/mail_manager.py

@@ -15,7 +15,7 @@ class MailManager:
 
     def send_mail(self, subject: str, content: str, receiver_email:str = "changsongd@126.com"):
         message = MIMEMultipart()
-        message['From'] = self.sender_email
+        message['From'] = f'Arlo <{self.sender_email}>'
         message['To'] = receiver_email
         message['Subject'] = subject
         message.attach(MIMEText(content, 'plain'))

models/__init__.py

@@ -0,0 +1,2 @@
+from models.source_content import SourceContent
+from models.article import Article

models/article.py

@@ -0,0 +1,50 @@
+from datetime import datetime
+from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy import String, Text, Integer, DateTime, func
+from models.base import Base
+
+
+class Article(Base):
+    __tablename__ = "t_article"
+
+    id: Mapped[int] = mapped_column(
+        Integer,
+        primary_key=True,
+        autoincrement=True,
+        comment="自动递增的唯一内容ID"
+    )
+
+    title: Mapped[str] = mapped_column(
+        String(256),
+        nullable=False,
+        index=True,
+        comment="标题"
+    )
+
+    keywords: Mapped[str | None] = mapped_column(
+        Text,
+        nullable=True,
+        comment="关键词"
+    )
+
+    content: Mapped[str | None] = mapped_column(
+        Text,
+        nullable=True,
+        comment="内容"
+    )
+
+    create_time: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        nullable=False,
+        comment="创建时间"
+    )
+
+    used: Mapped[bool] = mapped_column(
+        default=False,
+        nullable=False,
+        comment="是否已被使用"
+    )
+
+    def __repr__(self):
+        return f"<Article id={self.id} title={self.title!r} used={self.used!r}>"

models/base.py

@@ -0,0 +1,4 @@
+from sqlalchemy.orm import DeclarativeBase
+
+class Base(DeclarativeBase):
+    pass

models/source_content.py

@@ -0,0 +1,57 @@
+from datetime import datetime
+from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy import String, Text, Integer, DateTime, Index, func
+from models.base import Base
+
+
+class SourceContent(Base):
+    __tablename__ = "t_source_content"
+
+    id: Mapped[int] = mapped_column(
+        Integer,
+        primary_key=True,
+        autoincrement=True,
+        comment="自动递增的唯一内容ID"
+    )
+
+    link: Mapped[str] = mapped_column(
+        String(2048),
+        nullable=False,
+        index=True,
+        comment="链接"
+    )
+
+    platform: Mapped[str] = mapped_column(
+        String(32),
+        nullable=False,
+        comment="平台"
+    )
+
+    content: Mapped[str | None] = mapped_column(
+        Text,
+        nullable=True,
+        comment="内容"
+    )
+
+    create_time: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        nullable=False,
+        comment="创建时间"
+    )
+
+    update_time: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        onupdate=func.now(),
+        nullable=False,
+        comment="更新时间"
+    )
+
+    # ——可选优化：添加 项目 + 主题 联合唯一索引——
+    __table_args__ = (
+        Index("link", "link", unique=True),
+    )
+
+    def __repr__(self):
+        return f"<SourceContent id={self.id} link={self.link!r} platform={self.platform!r}>"

task/common_mail/common_mail_task.py

@@ -0,0 +1,24 @@
+from database.tscheduler.model import TScheduler
+from mail.mail_manager import send_mail
+from config.database import SessionLocal
+from models import Article
+from utils import logger
+
+
+def common_mail_task(scheduler: TScheduler):
+    with SessionLocal() as db:
+        # 获取需要发送的内容列表
+        articles = db.query(Article).filter(Article.used == False).all()
+        # 发送邮件
+        for article in articles:
+            subject = article.title
+            content = article.content
+            send_mail(subject, content, receiver_email="changsongd@126.com")
+            logger.info(f"send mail success with title {subject}, content {content[:20]}.")
+        # 更新数据库
+        for article in articles:
+            article.used = True
+        db.commit()
+
+if __name__ == '__main__':
+    common_mail_task(TScheduler())

task/manager_task.py

@@ -65,6 +65,30 @@ def execute_task(task: callable):
 
 # 从数据库加载任务
 def load_tasks(scheduler: BlockingScheduler):
+    def build_db_signature(db_task) -> str:
+        t = (db_task.trigger or "").lower()
+        if t == "interval":
+            sec = int(db_task.interval_seconds or 0)
+            return f"{db_task.task_name}#interval:seconds={sec}"
+        if t == "cron":
+            expr = (db_task.cron_expression or "").strip()
+            parts = expr.split()
+            # 规范化：确保 7 个字段，替换 Quartz 的 '?' 为 '*'
+            if len(parts) == 7:
+                if parts[5] == '?':
+                    parts[5] = '*'
+                expr = " ".join(parts)
+            return f"{db_task.task_name}#cron:{expr}"
+        if t == "date":
+            run_date = db_task.execution_date
+            if isinstance(run_date, datetime):
+                return f"{db_task.task_name}#date:{run_date.isoformat()}"
+            try:
+                return f"{db_task.task_name}#date:{str(run_date)}"
+            except Exception:
+                return f"{db_task.task_name}#date:invalid"
+        return f"{db_task.task_name}#unknown:{t}"
+
     with get_session() as db:
         tasks = get_tasks_by_executor(db, config.scheduler_name)
         for db_task in tasks:
@@ -86,6 +110,15 @@ def load_tasks(scheduler: BlockingScheduler):
             task_callable = partial(func, db_task)
             task_function = partial(execute_task, task_callable)
 
+            # 计算数据库中当前任务的签名
+            signature = build_db_signature(db_task)
+
+            job = scheduler.get_job(str(task_id))
+            # 如果任务已经存在并且签名相同，则跳过更新
+            if job and getattr(job, "name", None) == signature:
+                log(f"Task {db_task.task_name} (id={task_id}) already scheduled and unchanged, skipping.")
+                continue
+
             try:
                 if trigger == "interval":
                     scheduler.add_job(
@@ -93,7 +126,8 @@ def load_tasks(scheduler: BlockingScheduler):
                         "interval",
                         seconds=interval_seconds or 0,
                         id=str(task_id),
-                        replace_existing=True
+                        replace_existing=True,
+                        name=signature
                     )
                     log(f"Task {db_task.task_name} added/updated with interval {interval_seconds} seconds")
 
@@ -117,7 +151,8 @@ def load_tasks(scheduler: BlockingScheduler):
                         day_of_week=fields[5],
                         year=fields[6],
                         id=str(task_id),
-                        replace_existing=True
+                        replace_existing=True,
+                        name=signature
                     )
                     log(f"Task {db_task.task_name} added/updated with cron {db_task.cron_expression}")
 
@@ -139,7 +174,8 @@ def load_tasks(scheduler: BlockingScheduler):
                         "date",
                         run_date=run_date,
                         id=str(task_id),
-                        replace_existing=True
+                        replace_existing=True,
+                        name=signature
                     )
                     log(f"Task {db_task.task_name} added/updated with date {run_date}")
 

utils/__init__.py

@@ -0,0 +1 @@
+from utils.logger import logger

utils/logger.py

@@ -0,0 +1,37 @@
+import sys
+import os
+from loguru import logger
+from config.settings import settings
+
+# 移除默认的 handler（否则重复输出）
+logger.remove()
+
+if "console" in settings.LOG_TYPE:
+    # ======== 控制台输出 ========
+    logger.add(
+        sys.stdout,
+        level=settings.LOG_LEVEL,
+        format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> "
+            "| <level>{level: <8}</level> "
+            "| <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> "
+            "- <level>{message}</level>",
+    )
+
+if "file" in settings.LOG_TYPE:
+    # 日志目录
+    LOG_DIR = settings.LOG_FILE_PATH
+    if not os.path.exists(LOG_DIR):
+        os.makedirs(LOG_DIR)
+    
+    # ======== 文件输出（按天切割）========
+    logger.add(
+        f"{LOG_DIR}/app_{{time:YYYY-MM-DD}}.log",
+        rotation="00:00",        # 每天 0 点切割
+        retention="7 days",      # 保存 7 天
+        encoding="utf-8",
+        level=settings.LOG_LEVEL,
+        enqueue=True,            # 多线程安全
+        compression="zip",       # 自动压缩旧日志
+    )
+
+__all__ = ["logger"]