Init

Author: zxdong262

.github/workflows/linux.yml

@@ -0,0 +1,27 @@
+# This workflow will do a clean install of node dependencies, build the source code and run tests across different versions of node
+# For more information see: https://help.github.com/actions/language-and-framework-guides/using-nodejs-with-github-actions
+
+name: test
+
+on:
+  push:
+    branches: [ release, test ]
+
+jobs:
+  build:
+
+    runs-on: ${{ matrix.os }}
+    environment: build
+    if: "!contains(github.event.head_commit.message, 'skip ci') && !contains(github.event.head_commit.message, 'skip linux')"
+
+    strategy:
+      matrix:
+        os: [ubuntu-latest]
+
+    steps:
+    - uses: actions/checkout@v2
+
+    # script:
+    - run: cp sample.env .env
+    - run: bin/test
+

.gitignore

@@ -158,3 +158,4 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+/temp/

README.md

@@ -1,2 +1,42 @@
-# electerm-sync-server-python
-electerm-sync-server-python
+# Nodejs Electerm sync server
+
+[![Build Status](https://img.shields.io/endpoint.svg?url=https%3A%2F%2Factions-badge.atrox.dev%2Fatrox%2Fsync-dotenv%2Fbadge)](https://github.com/electerm/electerm-sync-server-python/actions)
+
+A simple electerm data sync server with python.
+
+## Use
+
+Requires python3
+
+```bash
+git clone git@github.com:electerm/electerm-sync-server-python.git
+cd electerm-sync-server-python
+pip install -r requirements.txt
+
+# create env file, then edit .env
+cp sample.env .env
+
+python src/app.py
+
+# would show something like
+# server running at http://127.0.0.1:7837
+
+# in electerm sync settings, set custom sync server with:
+# server url: http://127.0.0.1:7837
+# JWT_SECRET: your JWT_SECRET in .env
+# JWT_USER_NAME: one JWT_USER in .env
+```
+
+## Test
+
+```bash
+bin/test
+```
+
+## Write your own data store
+
+Just take [src/file-store.js](src/file-store.js) as an example, write your own read/write method
+
+## License
+
+MIT

bin/release

@@ -0,0 +1,10 @@
+#!/bin/bash
+cd `dirname $0`
+cd ..
+git co main
+git pull
+git pull
+git delete-branch release
+git create-branch release
+git push origin release -u
+git co -

bin/test

@@ -0,0 +1,4 @@
+#!/bin/bash
+cd `dirname $0`
+cd ..
+python tests/test.py

requirements.txt

@@ -0,0 +1,5 @@
+flask
+python-dotenv
+pylint
+flask_jwt_extended
+flask_jwt

sample.env

@@ -0,0 +1,10 @@
+# server port
+PORT=7837
+
+# server host
+HOST=127.0.0.1
+
+# jwt secret, make sure you change it in production
+JWT_SECRET=283hsdfye@!2@9oijnjSwda09
+JWT_USERS=username1,username2,xxxx,hhhh
+# FILE_STORE_PATH=/home/some/folder

src/app.py

@@ -0,0 +1,44 @@
+from flask import Flask, jsonify, request
+from flask_jwt_extended import JWTManager, jwt_required, get_jwt_identity
+from dotenv import load_dotenv
+from file_store import read, write
+import os
+
+load_dotenv()
+
+app = Flask(__name__)
+app.config['JWT_SECRET_KEY'] = os.environ['JWT_SECRET']
+app.config['JWT_IDENTITY_CLAIM'] = 'id'
+jwt = JWTManager(app)
+
+@app.route('/api/sync', methods=['GET', 'PUT'])
+@jwt_required()
+def sync():
+    user_id = get_jwt_identity()
+    users = os.environ['JWT_USERS'].split(',')
+    if user_id not in users:
+        return jsonify({'status': 'error', 'message': 'Unauthorized!'}), 401
+
+    if request.method == 'GET':
+        return read(user_id)
+
+    if request.method == 'PUT':
+        data = request.get_json()
+        return write(data, user_id)
+
+    return jsonify({'status': 'error', 'message': 'Unsupported method!'}), 405
+
+@app.route('/test', methods=['GET'])
+def test():
+    return 'ok'
+
+
+if __name__ == '__main__':
+  port = os.environ['PORT']
+  host = os.environ['HOST']
+  app.run(
+    host=host,
+    port=port,
+    debug=True,
+    load_dotenv=True
+  )

src/file_store.py

@@ -0,0 +1,20 @@
+import os
+import json
+
+file_dir = os.environ.get('FILE_STORE_PATH') or os.getcwd()
+
+def write(data, user_id):
+    json_body = json.dumps(data or {})
+    file_path = os.path.join(file_dir, f"{user_id}.json")
+    with open(file_path, 'w') as f:
+        f.write(json_body)
+    return "ok", 200
+
+def read(user_id):
+    file_path = os.path.join(file_dir, f"{user_id}.json")
+    if os.path.isfile(file_path):
+        with open(file_path, 'r') as f:
+            file_data = json.load(f)
+        return file_data, 200
+    else:
+        return "File not found", 404

tests/test.py

@@ -0,0 +1,60 @@
+import unittest
+import json
+import tempfile
+import os
+import sys
+from dotenv import load_dotenv
+
+load_dotenv()
+sys.path.append(os.path.dirname(os.path.abspath(__file__)) + "/..")
+
+from src.file_store import read, write
+
+class TestExample(unittest.TestCase):
+
+    def setUp(self):
+        self.test_user_id = 'test_user_id'
+        self.test_data = {'key': 'value'}
+
+    def test_write(self):
+        # Use temporary file for testing
+        with tempfile.NamedTemporaryFile() as f:
+
+            # Write to file
+            write_status, write_code = write(self.test_data, self.test_user_id)
+
+            # Test status and code
+            self.assertEqual(write_status, 'ok')
+            self.assertEqual(write_code, 200)
+
+            # Read from file
+            read_data, read_code = read(self.test_user_id)
+
+            # Test read data and code
+            self.assertEqual(read_data, self.test_data)
+            self.assertEqual(read_code, 200)
+
+    def test_read_not_found(self):
+        # Use temporary file for testing
+        with tempfile.TemporaryDirectory() as tempdir:
+            os.environ['FILE_STORE_PATH'] = tempdir
+
+            # Read from unexisting file
+            read_data, read_code = read(self.test_user_id + '0000')
+
+            # Test not found data and code
+            self.assertEqual(read_data, 'File not found')
+            self.assertEqual(read_code, 404)
+
+    def test_read(self):
+        # Use temporary file for testing
+        with tempfile.TemporaryDirectory() as tempdir:
+            # Read from unexisting file
+            read_data, read_code = read(self.test_user_id)
+
+            # Test not found data and code
+            self.assertEqual(read_data, self.test_data)
+            self.assertEqual(read_code, 200)
+
+if __name__ == '__main__':
+    unittest.main()