# coding:utf-8 from odps.udf import annotate from odps.distcache import get_cache_file @annotate("string,bigint->string") class fenci_str2str(object): def __init__(self): import sys sys.path.insert(0, './work/jieba-0.42.1-py3-none-any.zip') cache_file = get_cache_file('top_tags.txt') kv = {} for index, line in enumerate(cache_file): if index == 10000: break line = line.strip() if not line: continue k, _ = line.split('\t') kv[k] = index cache_file.close() self.kv = kv def evaluate(self, title, top_size): import jieba.analyse if title is None or len(title) == 0: return "" keys = jieba.analyse.extract_tags(title, topK=top_size, withWeight=False, allowPOS=('n', 'v')) if keys is None or len(keys) == 0: return "" keys_filter = [] for k in keys: if k in self.kv.keys(): keys_filter.append(k) if len(keys_filter) == 0: return "" return ",".join(keys_filter) @annotate("string,bigint,bigint->string") class get_top_tags(object): def evaluate(self, tag_ts, ts_limit, top): k_v = [kv.split(":") for kv in tag_ts.split(",")] result = dict() for tag, ts in k_v: ts = int(ts) if ts >= 0 and ts < ts_limit: tmp = result[tag] if tag in result else [0, -1] result[tag] = [tmp[0] + 1, max(tmp[1], ts)] if len(result) == 0: return "" sorted_result = sorted(result.items(), key=lambda x: (-x[1][0], -x[1][1]), reverse=False) top_keys = [item[0] for item in sorted_result[:top]] return ",".join(top_keys) @annotate("string,bigint,bigint->string") class get_top_tags_v2(object): def evaluate(self, tag_ts, ts_limit, top): k_v = [kv.split(":") for kv in tag_ts.split(",")] result = dict() for tag, ts, uv in k_v: ts = int(ts) uv = int(uv) if ts >= 0 and ts < ts_limit: tmp = result[tag] if tag in result else [0, -1] result[tag] = [tmp[0] + uv, max(tmp[1], ts)] if len(result) == 0: return "" sorted_result = sorted(result.items(), key=lambda x: (-x[1][0], -x[1][1]), reverse=False) top_keys = [item[0] for item in sorted_result[:top]] return ",".join(top_keys) @annotate("string,bigint,bigint->string") class mid_ts_dedup(object): def evaluate(self, mid_ts, ts_exp, time_diff): l = [(s.split(":")[0], s.split(":")[1]) for s in mid_ts.split(",")] m = {} for id, ts in l: ts = int(ts) if ts - ts_exp > time_diff or ts - ts_exp < 0: continue if id not in m or m[id] > ts: m[id] = ts ll = [] for k, v in m.items(): ll.append(k + ":" + str(v)) if len(ll) == 0: return "" res = ",".join(ll) return res @annotate("string->string") class get_cate1(object): def __init__(self): self.cate_list = [ "音乐", "剧情", "舞蹈", "动物", "三农", "科技", "财经", "母婴", "法律", "科普", "情感", "文化", "搞笑", "名人", "体育", "医疗健康", "时政", "奇人异象", "历史", "军事", "艺术", "美食", "旅行", "地域本地", "生活记录", "生活家居", "二次元", "游戏", "公益", "随拍", "职场", "教育", "摄影摄像", "时尚", "综艺", "电影", "电视剧", "汽车", "宗教", "短剧", "收藏品"] def evaluate(self, cate_str): if cate_str is None or cate_str == "" or cate_str == "无": return "unknown" result = [] for cate in self.cate_list: if cate in cate_str: result.append(cate) if len(result) == 0: result.append("unknown_" + cate_str) res = ",".join(result) return res @annotate("string->string") class get_cate2_all(object): def __init__(self): self.cate_list = [ "祝福音乐", "中国战争史", "中国历史影像", "知识科普", "正能量剧情", "杂技柔术", "早中晚好", "益智解密", "饮食健康", "戏曲戏剧", "未来科幻", "天气变化", "他国政策", "贪污腐败", "书法", "食品安全", "社会风气", "生活小妙招", "生活技巧科普", "省份城市亮点", "人生忠告", "人财诈骗", "亲子日常", "亲情音乐", "木工", "魔术特效", "迷信祝福", "民族异域音乐", "民生政策", "名画赏析", "美食教程", "麻将", "旅行攻略", "历史名人", "老综艺影像", "老年相关法律科普", "老年时尚", "老年审美美女", "老年生活", "老明星", "惊奇事件", "节日祝福", "健身操", "健康知识", "惠民新闻", "绘画", "怀念时光", "红歌老歌", "罕见画面", "国际文化", "国家统一", "国家力量", "国家科技力量", "搞笑段子", "风景实拍", "对口型表演", "动物萌宠", "动物表演", "大型集体艺术", "当代正能量人物", "传统文化", "吃播探店", "长寿知识", "本地生活", "K12教育", "(老)电影切片" ] def evaluate(self, cate_str): result = [] for cate in self.cate_list: if cate in cate_str: result.append(cate) if len(result) == 0: result.append("unknown") res = ",".join(result) return res @annotate("string->string") class get_cate2_only(object): def __init__(self): self.cate_list = [ "祝福音乐", "中国战争史", "中国历史影像", "知识科普", "正能量剧情", "杂技柔术", "早中晚好", "益智解密", "饮食健康", "戏曲戏剧", "未来科幻", "天气变化", "他国政策", "贪污腐败", "书法", "食品安全", "社会风气", "生活小妙招", "生活技巧科普", "省份城市亮点", "人生忠告", "人财诈骗", "亲子日常", "亲情音乐", "木工", "魔术特效", "迷信祝福", "民族异域音乐", "民生政策", "名画赏析", "美食教程", "麻将", "旅行攻略", "历史名人", "老综艺影像", "老年相关法律科普", "老年时尚", "老年审美美女", "老年生活", "老明星", "惊奇事件", "节日祝福", "健身操", "健康知识", "惠民新闻", "绘画", "怀念时光", "红歌老歌", "罕见画面", "国际文化", "国家统一", "国家力量", "国家科技力量", "搞笑段子", "风景实拍", "对口型表演", "动物萌宠", "动物表演", "大型集体艺术", "当代正能量人物", "传统文化", "吃播探店", "长寿知识", "本地生活", "K12教育", "(老)电影切片" ] def evaluate(self, cate_str): result = "unknown" for cate in self.cate_list: if cate in cate_str: result = cate break if result == "unknown": result = cate_str.split("、")[0].replace("品类-", "").replace("品类", "").replace("分数", "").replace("一人", "") result = self.clean_text(result) if len(result) == 0: result = "unknown" return result def clean_text(self, input_text): """ 去除字符串中的标点符号，只保留汉字和英文字符。 参数: input_text (str): 输入的字符串。 返回: str: 处理后的字符串。 """ # 使用正则表达式匹配汉字和英文字符 import re cleaned_text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z]', '', input_text) return cleaned_text @annotate("string->string") class clean_text(object): def __init__(self): import re self.re = re def evaluate(self, input_text): if input_text is None: return "" cleaned_text = self.re.sub(r"[^\u4e00-\u9fa5a-zA-Z0-9]", "", input_text) return "" if cleaned_text is None else cleaned_text @annotate("string->string") class deduplication4list(object): def evaluate(self, input_text): if input_text is None: return None result = list(set(input_text.split(","))) return ",".join(result)