添加数据库插入和更新功能

database/dao/mysql_dao.py

@@ -24,6 +24,7 @@ class MySqlDao:
         self._mock_order_tablename = "yunfu_mock_data"
         self._shopping_tablename = "tads_brandcul_cust_info_lbs_f"
         # self._shopping_tablename = "yunfu_shopping_mock_data"
+        self._report_tablename = "tads_brandcul_report"
         
         self._initialized = True
         

database/db/mysql.py

@@ -102,6 +102,62 @@ class MySqlDatabaseHelper:
         finally:
             session.close()
             
+    def insert_data(self, table_name, data_dict):
+        """插入单条数据到指定表"""
+        if not data_dict:
+            return 0
+        
+        columns = ", ".join(data_dict.keys())
+        values = ", ".join([f":{key}" for key in data_dict.keys()])
+        query = text(f"INSERT INTO {table_name} ({columns}) VALUES ({values})")
+        
+        session = self._DBSession()
+        
+        try:
+            result = session.execute(query, data_dict)
+            session.commit()
+            return result.rowcount
+        
+        except SQLAlchemyError as e:
+            session.rollback()
+            print(f"Error inserting data: {e}")
+            return 0
+        finally:
+            session.close()
+            
+    def update_data(self, table_name, update_dict, conditions, condition_params=None):
+        """更新表中符合条件的数据"""
+        if not update_dict:
+            return 0
+        
+        set_clause = ", ".join([f"{key} = :{key}" for key in update_dict.keys()])
+        
+        if len(conditions) == 1:
+            where_clause = f"WHERE {conditions[0]}"
+        elif len(conditions) > 1:
+            where_clause = f"WHERE {' AND '.join(conditions)}"
+        else:
+            where_clause = ""
+        
+        query = text(f"UPDATE {table_name} SET {set_clause} {where_clause}")
+        
+        params = update_dict.copy()
+        if condition_params:
+            params.update(condition_params)
+            
+        session = self._DBSession()
+        try:
+            result = session.execute(query, params)
+            session.commit()
+            return result.rowcount
+        except SQLAlchemyError as e:
+            session.rollback()
+            print(f"Error updating data: {e}")
+            return 0
+        
+        finally:
+            session.close()
+    
     def execute_query(self, query, params=None):
         """执行SQL语句 (无返回值, 如INSERT, UPDATE, DELETE)"""
         session = self._DBSession()
@@ -112,4 +168,33 @@ class MySqlDatabaseHelper:
             session.rollback()
             print(f"Error: {e}")
         finally:
-            session.close()
+            session.close()
+            
+if __name__ == '__main__':
+    db_helper = MySqlDatabaseHelper()
+    
+    table_name = 'tads_brandcul_report'
+    data_dict = {
+        'cultivacation_id': 10000002,
+        'city_uuid': '00000000000000000000000011445301',
+        'limit_cycle_name': '202505W1(05.05-05.11)',
+        'product_code': '440298',
+        'product_info_table': 'D72E3FAE8DCE4270BD23C3EC015C0A35',
+        'relation_table': 'AD889019FD4F4EE7B887981162BA09EC',
+        'similarity_product_table': 'CE436AC24D96461FA0C091CB01E9BC05',
+        'recommend_table': 'A7C5918B8DDB4BEA9D921936955CBAF6',
+    }
+    
+    # db_helper.insert_data(table_name, data_dict)
+    
+    update_data = {"val_table": "A7C5918B8DDB4BEA9D921936955CBAF6"}
+    conditions = [
+        "cultivacation_id = :cultivacation_id",
+        "city_uuid = :city_uuid"
+    ]
+    condition_params = {
+        'cultivacation_id': 10000001,
+        'city_uuid': '00000000000000000000000011445301',
+    }
+    
+    db_helper.update_data(table_name, update_data, conditions, condition_params)