做网站ps文字有锯齿,宣传片拍摄手法有哪些,网站模板怎么用法,网站开发逻辑图说明
UCS对象是基于GFGoLite进行封装#xff0c;且侧重于实现UCS规范。
内容
1 函数
我发现pydantic真是一个特别好用的东西#xff0c;可以确保在数据传递时的可靠#xff0c;以及对某个数据模型的描述。
以下#xff0c;UCS给出了id、time相关的brick映射#xff0…说明
UCS对象是基于GFGoLite进行封装且侧重于实现UCS规范。
内容
1 函数
我发现pydantic真是一个特别好用的东西可以确保在数据传递时的可靠以及对某个数据模型的描述。
以下UCS给出了id、time相关的brick映射并给出了时间到字符的转换。其方便之处在于本地几乎不需要管怎么实现的只要有服务支持即可。效率也足够高。
from typing import List, Optional
from pydantic import BaseModelimport requests as req
class UCS(BaseModel):__version__ 1.1gfgo_lite_server: str http://172.17.0.1:24090/def get_brick_name(self, some_id None):some_dict {}some_dict[rec_id] some_idurl self.gfgo_lite_server get_brick_name/res req.post(url, json some_dict).json()return res def get_brick_name_s(self, some_id_list None):some_dict {}some_dict[rec_id_list] some_id_listurl self.gfgo_lite_server get_brick_name_s/res req.post(url, json some_dict).json()return res def gfgo(self, kwargs None, pack_func None):url self.gfgo_lite_server gfgo/return req.post(url, json {kwargs:kwargs,pack_func:pack_func}).json()def dt_str2num(self, some_dt_str None):some_dict {}some_dict[some_dt_str] some_dt_strurl self.gfgo_lite_server str2num/res req.post(url, json some_dict).json()return resdef dt_str2num_s(self, some_dt_str_list None):some_dict {}some_dict[some_dt_str_list] some_dt_str_listurl self.gfgo_lite_server str2num_s/res req.post(url, json some_dict).json()return res# 时间名称def get_time_brick_name(self, dt_str_or_ts None):some_dict {}some_dict[dt_str_or_ts] dt_str_or_tsurl self.gfgo_lite_server get_time_brick_name/res req.post(url, json some_dict).json()return res def get_time_brick_name_s(self, dt_str_or_ts_list None):some_dict {}some_dict[dt_str_or_ts_list] dt_str_or_ts_listurl self.gfgo_lite_server get_time_brick_name_s/res req.post(url, json some_dict).json()return res # 1.1 def get_brick_list(self, start_brick_name None, end_brick_name None):some_dict {}some_dict[start_brick_name] start_brick_namesome_dict[end_brick_name] end_brick_nameurl self.gfgo_lite_server get_brick_list/res req.post(url, json some_dict).json()return res def get_brick_bounds(self, brick_name None):some_dict {}some_dict[brick_name] brick_nameurl self.gfgo_lite_server get_brick_bounds/res req.post(url, json some_dict).json()return res # 1.2 时间# get_time_brick_bounds_s - 待def get_time_brick_bounds_s(self, brick_name_list None, char_or_num char):some_dict {}some_dict[brick_name_list] brick_name_listsome_dict[char_or_num] char_or_numurl self.gfgo_lite_server get_time_brick_bounds_s/res req.post(url, json some_dict).json()return res # get_time_brick_listdef get_time_brick_list(self,start_brick_name None, end_brick_name None):some_dict {}some_dict[start_brick_name] start_brick_namesome_dict[end_brick_name] end_brick_nameurl self.gfgo_lite_server get_time_brick_list/res req.post(url, json some_dict).json()return res
2 应用
2.1 时间转换
ucs UCS(gfgo_lite_server http://xxx:24090/)
ucs.dt_str2num(2024-01-31 11:11:11)
1706670671ucs.dt_str2num_s([2024-01-31 11:11:11,2024-02-01 11:11:11 ])
[1706670671, 1706757071]反函数可以用get_time_str1
In [19]: get_time_str1(1706670671)
Out[19]: 2024-01-31 11:11:112.2 UCS-time 一个简单的应用如下我需要制定一个静态数据同步计划计划需要随着时间轴按照时间块熟顺序向前推进。具体来说这个对于数据库迁移(合并)与回测都有关系。 tbrick1 ucs.get_time_brick_name(2024-01-31 11:11:11)
2024.01.31.11
tbrick2 ucs.get_time_brick_name(2024-02-10 11:11:11)
2024.02.10.11tbrick_list ucs.get_time_brick_list(start_brick_name tbrick1, end_brick_name tbrick2)
[2024.01.31.11, 2024.01.31.12, 2024.01.31.13, ...]ucs.get_time_brick_bounds_s(tbrick_list[:3])
[[2024-01-31 11:00:00, 2024-01-31 12:00:00],[2024-01-31 12:00:00, 2024-01-31 13:00:00],[2024-01-31 13:00:00, 2024-01-31 14:00:00]]btw, 在探索完UCS-time之后我发现大约因为时间观念比较深入人心其实是可以更自由的转换的
# 将一般字符串转为UCS 名称
def dt_str2ucs_blockname(some_dt_str):some_dt_str1 some_dt_str.replace(-,.).replace( ,.).replace(:,.)return ..join(some_dt_str1.split(.)[:4])
dt_str2ucs_blockname(2024-06-24 09:30:00)
2024.06.24.09这些函数也被纳入基础函数有时候可以起到辅助的作用,特别在使用GFGolite不那么方便的时候。
一个更具体的场景我觉得是具有通用性的描述如下
1 数据没有顺序id但是具备入库时间(create_time)2 数据入库时间是不连续的有一些时间段密集更多的时间可能空闲但也并不是一条没有。3 每次执行时worker总是能够有效推进一个brick
具体做法
p01: 将需要的时间块按序持久化为本地文件并将对应的bounds作为字典
p02: 获取当前时间的时间块
# 将标准时间字符串转为brick
def timestr2timeblock(some_time_str None):x1 some_time_str.replace(-,.).replace( ,.).replace(:,.)return ..join(x1.split(.)[:4])the_time_now get_time_str1()
the_time_brick timestr2timeblock(the_time_now)p03:从gb中获取上一次处理过的最新brick并锁定在时间轴的位置。 worker_buffer_space sp_worker.general
tier1 some_tier
tier2 ucs_time_brick_ordered.sniffer
prefix ..join([worker_buffer_space,tier1,tier2]) .# Load
time_brick_list from_pickle(time_brick_list)
time_bounds_dict from_pickle(time_bounds_dict)
gb GlobalBuffer()
watch_brick_pos time_brick_list.index(the_time_brick)p04:按照已处理之后当前观察时间之前的原则选出available_time_brick_list
latest_data_brick gb.getx(prefix last_brick_handled)
if latest_data_brick is None:start_pos 0
else:start_pos time_brick_list.index(latest_data_brick)available_time_brick_list time_brick_list[start_pos1:watch_brick_pos]
p05:按顺序遍历时间块直到处理到一个有效的brick才中断本次循环
# cur_data_brick available_time_brick_list[0]
for cur_data_brick in available_time_brick_list:print(cur_data_brick)cur_bound time_bounds_dict[cur_data_brick]print(cur_bound)resp 【获取待处理数据】if len(resp):res_df pd.DataFrame(resp, columns keep_cols)if ok_to_handle:【将结果分批删除】listofdict2 slice_list_by_batch2(xxx.to_dict(orientrecords), 1000)for some_listofdict in listofdict2:【更新已处理块信息】gb.setx(prefix last_brick_handled,cur_data_brick, persistTrue)break
通过循环或者定时任务唤起这样的worker就可以。
