Parse [TAGS]

api/s8_tags.py

@@ -6,7 +6,7 @@ TAG_TYPE_LINK = 'LINK'
 def get_tag_schema(name: str) -> dict[str, dict[str, Any]]:
     return { 't_type' : {'type': 'str' , 'optional': False , 'readonly': False},
              'id'     : {'type': 'str' , 'optional': False , 'readonly': False},
-             'tag'    : {'type': 'str' , 'optional': True , 'readonly': False},}
+             'tag'    : {'type': 'str' , 'optional': True  , 'readonly': False},}
 
 
 def get_tag(name: str, t_type: str, id: str) -> dict[str, Any]:
@@ -77,3 +77,44 @@ def set_tag_cache(name: str, cs: ChangeSet) -> SqlChangeSet:
 
 def set_tag(name: str, cs: ChangeSet) -> ChangeSet:
     return execute_command(name, set_tag_cache(name, cs))
+
+
+class InpTag:
+    def __init__(self, line: str) -> None:
+        tokens = line.split()
+
+        num = len(tokens)
+        has_desc = tokens[-1].startswith(';')
+        num_without_desc = (num - 1) if has_desc else num
+
+        self.t_type = str(tokens[0])
+        self.id = str(tokens[1])
+        self.tag = str(tokens[2])
+
+
+def inp_in_tag(section: list[str]) -> ChangeSet:
+    cs = ChangeSet()
+    for s in section:
+        # skip comment
+        if s.startswith(';'):
+            continue
+        obj = InpTag(s)
+        cs.append({'operation': API_UPDATE, 'type': 'tag', 't_type': obj.t_type, 'id': obj.id, 'tag': obj.tag})
+    return cs
+
+
+def inp_out_tag(name: str) -> list[str]:
+    lines = []
+    objs = read_all(name, 'select * from tags_node')
+    for obj in objs:
+        t_type = TAG_TYPE_NODE
+        id = obj['id']
+        tag = obj['tag']
+        lines.append(f'{t_type} {id} {tag}')
+    objs = read_all(name, 'select * from tags_link')
+    for obj in objs:
+        t_type = TAG_TYPE_LINK
+        id = obj['id']
+        tag = obj['tag']
+        lines.append(f'{t_type} {id} {tag}')
+    return lines