add contact 2024-03-6

unify_api/modules/qkadmin/components/contact_us_cps.py

+# -*- coding:utf-8 -*-
+"""
+DATE：2024/3/7 11:00
+
+"""
+from dataclasses import dataclass
+from pot_libs.sanic_api import Model
+from pot_libs.sanic_api.column import Opt, Int, Str, List
+
+
+@dataclass
+class ContactsReq(Model):
+    page_num: int = Int("页码").eg(1)
+    page_size: int = Opt(Int("条数").eg(10))
+
+
+@dataclass
+class ContactInfo(Model):
+    id: int = Opt(Int("ID").eg(1))
+    user_name: str = Opt(Str("姓名").eg("ZZH"))
+    tele: str = Opt(Str("电话号码").eg("10000"))
+    email: str = Opt(Str("电子邮箱").eg("hu.jiayi@qkupower.com"))
+    company: str = Opt(Str("公司名称").eg("清科优能"))
+    msg: int = Opt(Str("留言").eg("来！一个电话、一个扫码，一个全新的合作机会！"))
+    create_time: str = Str("时间").eg("2024-12-29 18:06")
+
+
+@dataclass
+class ContactsRsp(Model):
+    total: int = Int("联系总数").eg(99)
+    contacts: list = List("联系列表").items(ContactInfo)

unify_api/modules/qkadmin/service/contact_us_srv.py

+# -*- coding:utf-8 -*-
+"""
+DATE：2024/3/7 11:08
+
+"""
+from pot_libs.mysql_util.mysql_util import MysqlUtil
+
+
+async def load_contact_records(page_num, page_size):
+    offsets = 0 if page_num == 0 else (page_num - 1) * page_size
+    async with MysqlUtil(db="official_web") as conn:
+        sql = "SELECT count(*) FROM official_web.contact_record;"
+        total = await conn.fetch_value(sql)
+
+        sql = f"SELECT * FROM official_web.contact_record " \
+              f"ORDER BY create_time DESC, id LIMIT %s OFFSET %s;"
+
+        records = await conn.fetchall(sql, (page_size, offsets))
+        return total, records

unify_api/modules/qkadmin/views/contact_us.py

+# -*- coding:utf-8 -*-
+"""
+DATE：2024/3/7 10:55
+
+"""
+from pot_libs.sanic_api import summary
+from unify_api.modules.qkadmin.components.contact_us_cps import (
+    ContactsReq, ContactsRsp, ContactInfo
+)
+from unify_api.modules.qkadmin.service.contact_us_srv import (
+    load_contact_records
+)
+
+
+@summary("联系我们记录")
+async def post_contact_records(request, body: ContactsReq) -> ContactsRsp:
+    page_num = int(body.page_num)
+    page_size = int(body.page_size)
+    total, records = await load_contact_records(page_num, page_size)
+    contacts = [ContactInfo(id=r["id"], user_name=r["user_name"],
+                            tele=r["tele"], company=r["company"],
+                            msg=r["msg"], create_time=str(r["create_time"]),
+                            email=r["email"]) for r in records]
+    return ContactsRsp(total=total, contacts=contacts)