release: 1.11.0

resolve #114
fix #124

CHANGELOG.md

@@ -1,5 +1,12 @@
 # 更新日志
 
+## 1.11.0
+
+- 改善 HLS 标准录制模式的稳定性
+- 兼容禁用弹幕的直播间
+- 支持 Bark 通知
+- 日志文件改为按天分割
+
 ## 1.10.0
 
 - 设置 umask 为 000 以确保创建的文件夹权限为 777

setup.cfg

@@ -53,6 +53,7 @@ install_requires =
     lxml >= 4.6.4, < 5.0.0
     toml >= 0.10.2, < 0.11.0
     m3u8 >= 1.0.0, < 2.0.0
+    av >= 10.0.0, < 11.0.0
     jsonpath == 0.82
     psutil >= 5.8.0, < 6.0.0
     reactivex >= 4.0.0, < 5.0.0

src/blrec/__init__.py

@@ -1,3 +1,3 @@
 __prog__ = 'blrec'
-__version__ = '1.10.0'
+__version__ = '1.11.0'
 __github__ = 'https://github.com/acgnhiki/blrec'

src/blrec/application.py

@@ -33,6 +33,7 @@ from .notification import (
     PushdeerNotifier,
     PushplusNotifier,
     TelegramNotifier,
+    BarkNotifier,
 )
 from .webhook import WebHookEmitter
 
@@ -331,11 +332,13 @@ class Application:
         self._pushdeer_notifier = PushdeerNotifier()
         self._pushplus_notifier = PushplusNotifier()
         self._telegram_notifier = TelegramNotifier()
+        self._bark_notifier = BarkNotifier()
         self._settings_manager.apply_email_notification_settings()
         self._settings_manager.apply_serverchan_notification_settings()
         self._settings_manager.apply_pushdeer_notification_settings()
         self._settings_manager.apply_pushplus_notification_settings()
         self._settings_manager.apply_telegram_notification_settings()
+        self._settings_manager.apply_bark_notification_settings()
 
     def _setup_webhooks(self) -> None:
         self._webhook_emitter = WebHookEmitter()
@@ -367,11 +370,13 @@ class Application:
         self._pushdeer_notifier.disable()
         self._pushplus_notifier.disable()
         self._telegram_notifier.disable()
+        self._bark_notifier.disable()
         del self._email_notifier
         del self._serverchan_notifier
         del self._pushdeer_notifier
         del self._pushplus_notifier
         del self._telegram_notifier
+        del self._bark_notifier
 
     def _destroy_webhooks(self) -> None:
         self._webhook_emitter.disable()

src/blrec/bili/danmaku_client.py

@@ -267,6 +267,9 @@ class DanmakuClient(EventEmitter[DanmakuListener], AsyncStoppableMixin):
         while True:
             try:
                 wsmsg = await self._ws.receive(timeout=self._HEARTBEAT_INTERVAL)
+            except asyncio.TimeoutError as e:
+                logger.debug(f'Failed to receive message due to: {repr(e)}')
+                continue
             except Exception as e:
                 await self._handle_error(e)
             else:

src/blrec/core/hls_raw_stream_recorder_impl.py

@@ -48,7 +48,9 @@ class HLSRawStreamRecorderImpl(StreamRecorderImpl):
 
         self._playlist_fetcher = hls_ops.PlaylistFetcher(self._live, self._session)
         self._playlist_dumper = hls_ops.PlaylistDumper(self._path_provider)
-        self._segment_fetcher = hls_ops.SegmentFetcher(self._live, self._session)
+        self._segment_fetcher = hls_ops.SegmentFetcher(
+            self._live, self._session, self._stream_url_resolver
+        )
         self._segment_dumper = hls_ops.SegmentDumper(self._playlist_dumper)
         self._ff_metadata_dumper = MetadataDumper(
             self._playlist_dumper, self._metadata_provider

src/blrec/core/hls_stream_recorder_impl.py

@@ -48,16 +48,16 @@ class HLSStreamRecorderImpl(StreamRecorderImpl):
         )
 
         self._playlist_fetcher = hls_ops.PlaylistFetcher(self._live, self._session)
-        self._playlist_resolver = hls_ops.PlaylistResolver()
-        self._segment_fetcher = hls_ops.SegmentFetcher(self._live, self._session)
+        self._playlist_resolver = hls_ops.PlaylistResolver(self._stream_url_resolver)
+        self._segment_fetcher = hls_ops.SegmentFetcher(
+            self._live, self._session, self._stream_url_resolver
+        )
         self._segment_remuxer = hls_ops.SegmentRemuxer(live)
 
         self._prober = hls_ops.Prober()
         self._dl_statistics = core_ops.SizedStatistics()
 
-        self._stream_parser = core_ops.StreamParser(
-            self._stream_param_holder, ignore_eof=True, ignore_value_error=True
-        )
+        self._segment_parser = hls_ops.SegmentParser()
         self._analyser = flv_ops.Analyser()
         self._injector = flv_ops.Injector(self._metadata_provider)
         self._join_point_extractor = flv_ops.JoinPointExtractor()
@@ -144,14 +144,11 @@ class HLSStreamRecorderImpl(StreamRecorderImpl):
                 self._segment_fetcher,
                 self._dl_statistics,
                 self._prober,
-                ops.observe_on(
-                    NewThreadScheduler(self._thread_factory('SegmentRemuxer'))
-                ),
-                self._segment_remuxer,
                 ops.observe_on(
                     NewThreadScheduler(self._thread_factory('StreamRecorder'))
                 ),
-                self._stream_parser,
+                self._segment_remuxer,
+                self._segment_parser,
                 flv_ops.process(),
                 self._cutter,
                 self._limiter,

src/blrec/core/operators/stream_url_resolver.py

@@ -5,7 +5,6 @@ from typing import Optional
 from urllib.parse import urlparse
 
 import requests
-import urllib3
 from reactivex import Observable, abc
 from reactivex import operators as ops
 
@@ -29,7 +28,6 @@ __all__ = ('StreamURLResolver',)
 
 
 logger = logging.getLogger(__name__)
-logging.getLogger(urllib3.__name__).setLevel(logging.WARNING)
 
 
 class StreamURLResolver(AsyncCooperationMixin):
@@ -49,14 +47,14 @@ class StreamURLResolver(AsyncCooperationMixin):
     def stream_host(self) -> str:
         return self._stream_host
 
-    def _reset(self) -> None:
+    def reset(self) -> None:
         self._stream_url = ''
         self._stream_host = ''
         self._stream_params = None
 
     def __call__(self, source: Observable[StreamParams]) -> Observable[str]:
-        self._reset()
-        return self._solve(source).pipe(
+        self.reset()
+        return self._solve(source).pipe(  # type: ignore
             ops.do_action(on_error=self._before_retry),
             utils_ops.retry(delay=1, should_retry=self._should_retry),
         )

src/blrec/core/stream_recorder_impl.py

@@ -1,3 +1,4 @@
+import io
 import logging
 from abc import ABC, abstractmethod
 from datetime import datetime
@@ -75,8 +76,8 @@ class StreamRecorderImpl(
         self._session = requests.Session()
 
         self._recording_mode = recording_mode
-        self._buffer_size = buffer_size
-        self._read_timeout = read_timeout
+        self._buffer_size = buffer_size or io.DEFAULT_BUFFER_SIZE
+        self._read_timeout = read_timeout or 3
         self._filesize_limit = filesize_limit
         self._duration_limit = duration_limit
 

src/blrec/data/webapp/index.html

@@ -10,6 +10,6 @@
 <body>
   <app-root></app-root>
   <noscript>Please enable JavaScript to continue using this application.</noscript>
-<script src="runtime.c6818dbcd7b06106.js" type="module"></script><script src="polyfills.4b08448aee19bb22.js" type="module"></script><script src="main.6da8ea192405b948.js" type="module"></script>
+<script src="runtime.0560bf422fb1ab67.js" type="module"></script><script src="polyfills.4b08448aee19bb22.js" type="module"></script><script src="main.dbd09d2079405adc.js" type="module"></script>
 
 </body></html>

src/blrec/data/webapp/ngsw.json

@@ -1,6 +1,6 @@
 {
   "configVersion": 1,
-  "timestamp": 1661579095139,
+  "timestamp": 1667707262978,
   "index": "/index.html",
   "assetGroups": [
     {
@@ -14,15 +14,15 @@
         "/103.5b5d2a6e5a8a7479.js",
         "/146.5a8902910bda9e87.js",
         "/183.ee55fc76717674c3.js",
-        "/205.cf2caa9b46b14212.js",
         "/45.c90c3cea2bf1a66e.js",
-        "/91.cab8652a2fa56b1a.js",
+        "/548.91bbb60199d9e944.js",
+        "/91.3c224fe84835dadd.js",
         "/common.858f777e9296e6f2.js",
         "/index.html",
-        "/main.6da8ea192405b948.js",
+        "/main.dbd09d2079405adc.js",
         "/manifest.webmanifest",
         "/polyfills.4b08448aee19bb22.js",
-        "/runtime.c6818dbcd7b06106.js",
+        "/runtime.0560bf422fb1ab67.js",
         "/styles.2e152d608221c2ee.css"
       ],
       "patterns": []
@@ -1637,9 +1637,9 @@
     "/103.5b5d2a6e5a8a7479.js": "cc0240f217015b6d4ddcc14f31fcc42e1c1c282a",
     "/146.5a8902910bda9e87.js": "d9c33c7073662699f00f46f3a384ae5b749fdef9",
     "/183.ee55fc76717674c3.js": "2628c996ec80a6c6703d542d34ac95194283bcf8",
-    "/205.cf2caa9b46b14212.js": "749df896fbbd279dcf49318963f0ce074c5df87f",
     "/45.c90c3cea2bf1a66e.js": "e5bfb8cf3803593e6b8ea14c90b3d3cb6a066764",
-    "/91.cab8652a2fa56b1a.js": "c11ebf28472c8a75653f7b27b5cffdec477830fe",
+    "/548.91bbb60199d9e944.js": "062b3a6424284294e5774bcb08ca76df7b0c4216",
+    "/91.3c224fe84835dadd.js": "2e3cdb6c44a8cf3241fe8dd89b27c37f212768f8",
     "/assets/animal/panda.js": "fec2868bb3053dd2da45f96bbcb86d5116ed72b1",
     "/assets/animal/panda.svg": "bebd302cdc601e0ead3a6d2710acf8753f3d83b1",
     "/assets/fill/.gitkeep": "da39a3ee5e6b4b0d3255bfef95601890afd80709",
@@ -3234,11 +3234,11 @@
     "/assets/twotone/warning.js": "fb2d7ea232f3a99bf8f080dbc94c65699232ac01",
     "/assets/twotone/warning.svg": "8c7a2d3e765a2e7dd58ac674870c6655cecb0068",
     "/common.858f777e9296e6f2.js": "b68ca68e1e214a2537d96935c23410126cc564dd",
-    "/index.html": "80797fa46f33b7bcf402788a5d0d0516b77f23b1",
-    "/main.6da8ea192405b948.js": "b8995c7d8ccd465769b90936db5e0a337a827a58",
+    "/index.html": "17482e27906b5ae0447920edbf4bf4f4c0c1838b",
+    "/main.dbd09d2079405adc.js": "2f7284b616ed9fc433b612c9dca53dc06a0f3aa1",
     "/manifest.webmanifest": "62c1cb8c5ad2af551a956b97013ab55ce77dd586",
     "/polyfills.4b08448aee19bb22.js": "8e73f2d42cc13ca353cea5c886d930bd6da08d0d",
-    "/runtime.c6818dbcd7b06106.js": "00160f946c5d007a956f5f61293cbd3bed2756dc",
+    "/runtime.0560bf422fb1ab67.js": "74c07903a5fd6d43a0d7690a93b0927f8eead22c",
     "/styles.2e152d608221c2ee.css": "9830389a46daa5b4511e0dd343aad23ca9f9690f"
   },
   "navigationUrls": [

src/blrec/data/webapp/runtime.0560bf422fb1ab67.js

@@ -0,0 +1 @@
+(()=>{"use strict";var e,v={},m={};function r(e){var f=m[e];if(void 0!==f)return f.exports;var t=m[e]={exports:{}};return v[e].call(t.exports,t,t.exports,r),t.exports}r.m=v,e=[],r.O=(f,t,i,o)=>{if(!t){var a=1/0;for(n=0;n<e.length;n++){for(var[t,i,o]=e[n],c=!0,l=0;l<t.length;l++)(!1&o||a>=o)&&Object.keys(r.O).every(p=>r.O[p](t[l]))?t.splice(l--,1):(c=!1,o<a&&(a=o));if(c){e.splice(n--,1);var d=i();void 0!==d&&(f=d)}}return f}o=o||0;for(var n=e.length;n>0&&e[n-1][2]>o;n--)e[n]=e[n-1];e[n]=[t,i,o]},r.n=e=>{var f=e&&e.__esModule?()=>e.default:()=>e;return r.d(f,{a:f}),f},r.d=(e,f)=>{for(var t in f)r.o(f,t)&&!r.o(e,t)&&Object.defineProperty(e,t,{enumerable:!0,get:f[t]})},r.f={},r.e=e=>Promise.all(Object.keys(r.f).reduce((f,t)=>(r.f[t](e,f),f),[])),r.u=e=>(592===e?"common":e)+"."+{45:"c90c3cea2bf1a66e",91:"3c224fe84835dadd",103:"5b5d2a6e5a8a7479",146:"5a8902910bda9e87",183:"ee55fc76717674c3",548:"91bbb60199d9e944",592:"858f777e9296e6f2"}[e]+".js",r.miniCssF=e=>{},r.o=(e,f)=>Object.prototype.hasOwnProperty.call(e,f),(()=>{var e={},f="blrec:";r.l=(t,i,o,n)=>{if(e[t])e[t].push(i);else{var a,c;if(void 0!==o)for(var l=document.getElementsByTagName("script"),d=0;d<l.length;d++){var u=l[d];if(u.getAttribute("src")==t||u.getAttribute("data-webpack")==f+o){a=u;break}}a||(c=!0,(a=document.createElement("script")).type="module",a.charset="utf-8",a.timeout=120,r.nc&&a.setAttribute("nonce",r.nc),a.setAttribute("data-webpack",f+o),a.src=r.tu(t)),e[t]=[i];var s=(g,p)=>{a.onerror=a.onload=null,clearTimeout(b);var _=e[t];if(delete e[t],a.parentNode&&a.parentNode.removeChild(a),_&&_.forEach(h=>h(p)),g)return g(p)},b=setTimeout(s.bind(null,void 0,{type:"timeout",target:a}),12e4);a.onerror=s.bind(null,a.onerror),a.onload=s.bind(null,a.onload),c&&document.head.appendChild(a)}}})(),r.r=e=>{"undefined"!=typeof Symbol&&Symbol.toStringTag&&Object.defineProperty(e,Symbol.toStringTag,{value:"Module"}),Object.defineProperty(e,"__esModule",{value:!0})},(()=>{var e;r.tu=f=>(void 0===e&&(e={createScriptURL:t=>t},"undefined"!=typeof trustedTypes&&trustedTypes.createPolicy&&(e=trustedTypes.createPolicy("angular#bundler",e))),e.createScriptURL(f))})(),r.p="",(()=>{var e={666:0};r.f.j=(i,o)=>{var n=r.o(e,i)?e[i]:void 0;if(0!==n)if(n)o.push(n[2]);else if(666!=i){var a=new Promise((u,s)=>n=e[i]=[u,s]);o.push(n[2]=a);var c=r.p+r.u(i),l=new Error;r.l(c,u=>{if(r.o(e,i)&&(0!==(n=e[i])&&(e[i]=void 0),n)){var s=u&&("load"===u.type?"missing":u.type),b=u&&u.target&&u.target.src;l.message="Loading chunk "+i+" failed.\n("+s+": "+b+")",l.name="ChunkLoadError",l.type=s,l.request=b,n[1](l)}},"chunk-"+i,i)}else e[i]=0},r.O.j=i=>0===e[i];var f=(i,o)=>{var l,d,[n,a,c]=o,u=0;if(n.some(b=>0!==e[b])){for(l in a)r.o(a,l)&&(r.m[l]=a[l]);if(c)var s=c(r)}for(i&&i(o);u<n.length;u++)r.o(e,d=n[u])&&e[d]&&e[d][0](),e[n[u]]=0;return r.O(s)},t=self.webpackChunkblrec=self.webpackChunkblrec||[];t.forEach(f.bind(null,0)),t.push=f.bind(null,t.push.bind(t))})()})();

src/blrec/data/webapp/runtime.c6818dbcd7b06106.js

@@ -1 +0,0 @@
-(()=>{"use strict";var e,v={},m={};function r(e){var i=m[e];if(void 0!==i)return i.exports;var t=m[e]={exports:{}};return v[e].call(t.exports,t,t.exports,r),t.exports}r.m=v,e=[],r.O=(i,t,f,o)=>{if(!t){var a=1/0;for(n=0;n<e.length;n++){for(var[t,f,o]=e[n],c=!0,l=0;l<t.length;l++)(!1&o||a>=o)&&Object.keys(r.O).every(p=>r.O[p](t[l]))?t.splice(l--,1):(c=!1,o<a&&(a=o));if(c){e.splice(n--,1);var d=f();void 0!==d&&(i=d)}}return i}o=o||0;for(var n=e.length;n>0&&e[n-1][2]>o;n--)e[n]=e[n-1];e[n]=[t,f,o]},r.n=e=>{var i=e&&e.__esModule?()=>e.default:()=>e;return r.d(i,{a:i}),i},r.d=(e,i)=>{for(var t in i)r.o(i,t)&&!r.o(e,t)&&Object.defineProperty(e,t,{enumerable:!0,get:i[t]})},r.f={},r.e=e=>Promise.all(Object.keys(r.f).reduce((i,t)=>(r.f[t](e,i),i),[])),r.u=e=>(592===e?"common":e)+"."+{45:"c90c3cea2bf1a66e",91:"cab8652a2fa56b1a",103:"5b5d2a6e5a8a7479",146:"5a8902910bda9e87",183:"ee55fc76717674c3",205:"cf2caa9b46b14212",592:"858f777e9296e6f2"}[e]+".js",r.miniCssF=e=>{},r.o=(e,i)=>Object.prototype.hasOwnProperty.call(e,i),(()=>{var e={},i="blrec:";r.l=(t,f,o,n)=>{if(e[t])e[t].push(f);else{var a,c;if(void 0!==o)for(var l=document.getElementsByTagName("script"),d=0;d<l.length;d++){var u=l[d];if(u.getAttribute("src")==t||u.getAttribute("data-webpack")==i+o){a=u;break}}a||(c=!0,(a=document.createElement("script")).type="module",a.charset="utf-8",a.timeout=120,r.nc&&a.setAttribute("nonce",r.nc),a.setAttribute("data-webpack",i+o),a.src=r.tu(t)),e[t]=[f];var s=(g,p)=>{a.onerror=a.onload=null,clearTimeout(b);var _=e[t];if(delete e[t],a.parentNode&&a.parentNode.removeChild(a),_&&_.forEach(h=>h(p)),g)return g(p)},b=setTimeout(s.bind(null,void 0,{type:"timeout",target:a}),12e4);a.onerror=s.bind(null,a.onerror),a.onload=s.bind(null,a.onload),c&&document.head.appendChild(a)}}})(),r.r=e=>{"undefined"!=typeof Symbol&&Symbol.toStringTag&&Object.defineProperty(e,Symbol.toStringTag,{value:"Module"}),Object.defineProperty(e,"__esModule",{value:!0})},(()=>{var e;r.tu=i=>(void 0===e&&(e={createScriptURL:t=>t},"undefined"!=typeof trustedTypes&&trustedTypes.createPolicy&&(e=trustedTypes.createPolicy("angular#bundler",e))),e.createScriptURL(i))})(),r.p="",(()=>{var e={666:0};r.f.j=(f,o)=>{var n=r.o(e,f)?e[f]:void 0;if(0!==n)if(n)o.push(n[2]);else if(666!=f){var a=new Promise((u,s)=>n=e[f]=[u,s]);o.push(n[2]=a);var c=r.p+r.u(f),l=new Error;r.l(c,u=>{if(r.o(e,f)&&(0!==(n=e[f])&&(e[f]=void 0),n)){var s=u&&("load"===u.type?"missing":u.type),b=u&&u.target&&u.target.src;l.message="Loading chunk "+f+" failed.\n("+s+": "+b+")",l.name="ChunkLoadError",l.type=s,l.request=b,n[1](l)}},"chunk-"+f,f)}else e[f]=0},r.O.j=f=>0===e[f];var i=(f,o)=>{var l,d,[n,a,c]=o,u=0;if(n.some(b=>0!==e[b])){for(l in a)r.o(a,l)&&(r.m[l]=a[l]);if(c)var s=c(r)}for(f&&f(o);u<n.length;u++)r.o(e,d=n[u])&&e[d]&&e[d][0](),e[n[u]]=0;return r.O(s)},t=self.webpackChunkblrec=self.webpackChunkblrec||[];t.forEach(i.bind(null,0)),t.push=i.bind(null,t.push.bind(t))})()})();

src/blrec/flv/format.py

@@ -1,37 +1,33 @@
-from io import BytesIO, SEEK_CUR
+from io import SEEK_CUR, BytesIO
 from typing import cast
 
 import attr
 
-from .struct_io import StructReader, StructWriter
+from .exceptions import FlvDataError, FlvHeaderError, FlvTagError
 from .io_protocols import RandomIO
-from .exceptions import FlvHeaderError, FlvDataError, FlvTagError
 from .models import (
-    FlvTag,
-    TagType,
-    FlvHeader,
-    FlvTagHeader,
-    TAG_HEADER_SIZE,
     AUDIO_TAG_HEADER_SIZE,
+    TAG_HEADER_SIZE,
     VIDEO_TAG_HEADER_SIZE,
-
+    AACPacketType,
     AudioTag,
     AudioTagHeader,
+    AVCPacketType,
+    CodecID,
+    FlvHeader,
+    FlvTag,
+    FlvTagHeader,
+    FrameType,
+    ScriptTag,
     SoundFormat,
     SoundRate,
     SoundSize,
     SoundType,
-    AACPacketType,
-
+    TagType,
     VideoTag,
     VideoTagHeader,
-    CodecID,
-    FrameType,
-    AVCPacketType,
-
-    ScriptTag,
 )
-
+from .struct_io import StructReader, StructWriter
 
 __all__ = 'FlvParser', 'FlvDumper'
 
@@ -73,7 +69,7 @@ class FlvParser:
             body_size = tag_header.data_size - AUDIO_TAG_HEADER_SIZE
             if no_body:
                 self._stream.seek(body_size, SEEK_CUR)
-                body = None
+                body = b''
             else:
                 body = self._reader.read(body_size)
             audio_tag_header = self.parse_audio_tag_header(header_data)
@@ -88,7 +84,7 @@ class FlvParser:
             body_size = tag_header.data_size - VIDEO_TAG_HEADER_SIZE
             if no_body:
                 self._stream.seek(body_size, SEEK_CUR)
-                body = None
+                body = b''
             else:
                 body = self._reader.read(body_size)
             video_tag_header = self.parse_video_tag_header(header_data)
@@ -102,14 +98,10 @@ class FlvParser:
             body_size = tag_header.data_size
             if no_body:
                 self._stream.seek(body_size, SEEK_CUR)
-                body = None
+                body = b''
             else:
                 body = self._reader.read(body_size)
-            return ScriptTag(
-                offset=offset,
-                **attr.asdict(tag_header),
-                body=body,
-            )
+            return ScriptTag(offset=offset, **attr.asdict(tag_header), body=body)
         else:
             raise FlvDataError(f'Unsupported tag type: {tag_header.tag_type}')
 
@@ -157,9 +149,7 @@ class FlvParser:
         flag = reader.read_ui8()
         sound_format = SoundFormat(flag >> 4)
         if sound_format != SoundFormat.AAC:
-            raise FlvDataError(
-                f'Unsupported sound format: {sound_format}', data
-            )
+            raise FlvDataError(f'Unsupported sound format: {sound_format}', data)
         sound_rate = SoundRate((flag >> 2) & 0b0000_0011)
         sound_size = SoundSize((flag >> 1) & 0b0000_0001)
         sound_type = SoundType(flag & 0b0000_0001)
@@ -177,9 +167,7 @@ class FlvParser:
             raise FlvDataError(f'Unsupported video codec: {codec_id}', data)
         avc_packet_type = AVCPacketType(reader.read_ui8())
         composition_time = reader.read_ui24()
-        return VideoTagHeader(
-            frame_type, codec_id, avc_packet_type, composition_time
-        )
+        return VideoTagHeader(frame_type, codec_id, avc_packet_type, composition_time)
 
 
 class FlvDumper:
@@ -221,28 +209,24 @@ class FlvDumper:
     def dump_flv_tag_header(self, tag: FlvTag) -> None:
         self._writer.write_ui8((int(tag.filtered) << 5) | tag.tag_type.value)
         self._writer.write_ui24(tag.data_size)
-        self._writer.write_ui24(tag.timestamp & 0x00ffffff)
+        self._writer.write_ui24(tag.timestamp & 0x00FFFFFF)
         self._writer.write_ui8(tag.timestamp >> 24)
         self._writer.write_ui24(tag.stream_id)
 
     def dump_audio_tag_header(self, tag: AudioTag) -> None:
         if tag.sound_format != SoundFormat.AAC:
-            raise FlvDataError(
-                f'Unsupported sound format: {tag.sound_format}', tag
-            )
+            raise FlvDataError(f'Unsupported sound format: {tag.sound_format}', tag)
         self._writer.write_ui8(
-            (tag.sound_format.value << 4) |
-            (tag.sound_rate.value << 2) |
-            (tag.sound_size.value << 1) |
-            tag.sound_type.value
+            (tag.sound_format.value << 4)
+            | (tag.sound_rate.value << 2)
+            | (tag.sound_size.value << 1)
+            | tag.sound_type.value
         )
         self._writer.write_ui8(tag.aac_packet_type)
 
     def dump_video_tag_header(self, tag: VideoTag) -> None:
         if tag.codec_id != CodecID.AVC:
             raise FlvDataError(f'Unsupported video codec: {tag.codec_id}', tag)
-        self._writer.write_ui8(
-            (tag.frame_type.value << 4) | tag.codec_id.value
-        )
+        self._writer.write_ui8((tag.frame_type.value << 4) | tag.codec_id.value)
         self._writer.write_ui8(tag.avc_packet_type.value)
         self._writer.write_ui24(tag.composition_time)

src/blrec/hls/exceptions.py

@@ -1,2 +1,10 @@
 class SegmentDataCorrupted(ValueError):
     pass
+
+
+class NoNewSegments(Exception):
+    pass
+
+
+class FetchSegmentError(Exception):
+    pass

src/blrec/hls/operators/__init__.py

@@ -4,6 +4,7 @@ from .playlist_resolver import PlaylistResolver
 from .prober import Prober, StreamProfile
 from .segment_dumper import SegmentDumper
 from .segment_fetcher import InitSectionData, SegmentData, SegmentFetcher
+from .segment_parser import SegmentParser
 from .segment_remuxer import SegmentRemuxer
 
 __all__ = (
@@ -15,6 +16,7 @@ __all__ = (
     'SegmentData',
     'SegmentDumper',
     'SegmentFetcher',
+    'SegmentParser',
     'SegmentRemuxer',
     'StreamProfile',
 )

src/blrec/hls/operators/playlist_resolver.py

@@ -5,20 +5,30 @@ import os
 from typing import Optional
 
 import m3u8
-import urllib3
 from reactivex import Observable, abc
+from reactivex import operators as ops
 from reactivex.disposable import CompositeDisposable, Disposable, SerialDisposable
 
+from blrec.core import operators as core_ops
+from blrec.utils import operators as utils_ops
+
+from ..exceptions import NoNewSegments
+
 __all__ = ('PlaylistResolver',)
 
 
 logger = logging.getLogger(__name__)
-logging.getLogger(urllib3.__name__).setLevel(logging.WARNING)
 
 
 class PlaylistResolver:
+    def __init__(self, stream_url_resolver: core_ops.StreamURLResolver) -> None:
+        self._stream_url_resolver = stream_url_resolver
+
     def __call__(self, source: Observable[m3u8.M3U8]) -> Observable[m3u8.Segment]:
-        return self._solve(source)
+        return self._solve(source).pipe(
+            ops.do_action(on_error=self._before_retry),
+            utils_ops.retry(should_retry=self._should_retry),
+        )
 
     def _name_of(self, uri: str) -> str:
         name, ext = os.path.splitext(uri)
@@ -35,18 +45,18 @@ class PlaylistResolver:
             disposed = False
             subscription = SerialDisposable()
 
+            attempts: int = 0
             last_sequence_number: Optional[int] = None
 
             def on_next(playlist: m3u8.M3U8) -> None:
-                nonlocal last_sequence_number
+                nonlocal attempts, last_sequence_number
 
                 if playlist.is_endlist:
                     logger.debug('Playlist ended')
 
+                new_segments = []
                 for seg in playlist.segments:
-                    uri = seg.uri
-                    name = self._name_of(uri)
-                    num = int(name)
+                    num = self._sequence_number_of(seg.uri)
                     if last_sequence_number is not None:
                         if last_sequence_number >= num:
                             continue
@@ -57,9 +67,21 @@ class PlaylistResolver:
                                 f'current sequence number: {num}'
                             )
                             seg.discontinuity = True
-                    observer.on_next(seg)
+                    new_segments.append(seg)
                     last_sequence_number = num
 
+                if not new_segments:
+                    attempts += 1
+                    if attempts > 3:
+                        attempts = 0
+                        observer.on_error(NoNewSegments())
+                    return
+                else:
+                    attempts = 0
+
+                for seg in new_segments:
+                    observer.on_next(seg)
+
             def dispose() -> None:
                 nonlocal disposed
                 nonlocal last_sequence_number
@@ -73,3 +95,15 @@ class PlaylistResolver:
             return CompositeDisposable(subscription, Disposable(dispose))
 
         return Observable(subscribe)
+
+    def _should_retry(self, exc: Exception) -> bool:
+        if isinstance(exc, NoNewSegments):
+            return True
+        else:
+            return False
+
+    def _before_retry(self, exc: Exception) -> None:
+        if not isinstance(exc, NoNewSegments):
+            return
+        logger.warning('No new segments received, trying to update the stream url.')
+        self._stream_url_resolver.reset()

src/blrec/hls/operators/segment_fetcher.py

@@ -10,13 +10,23 @@ import requests
 import urllib3
 from m3u8.model import InitializationSection
 from reactivex import Observable, abc
+from reactivex import operators as ops
 from reactivex.disposable import CompositeDisposable, Disposable, SerialDisposable
-from tenacity import retry, retry_if_exception_type, stop_after_delay, wait_exponential
+from tenacity import (
+    retry,
+    retry_all,
+    retry_if_exception_type,
+    retry_if_not_exception_type,
+    stop_after_delay,
+    wait_exponential,
+)
 
 from blrec.bili.live import Live
+from blrec.core import operators as core_ops
+from blrec.utils import operators as utils_ops
 from blrec.utils.hash import cksum
 
-from ..exceptions import SegmentDataCorrupted
+from ..exceptions import FetchSegmentError
 
 __all__ = ('SegmentFetcher', 'InitSectionData', 'SegmentData')
 
@@ -43,14 +53,23 @@ class SegmentData:
 
 
 class SegmentFetcher:
-    def __init__(self, live: Live, session: requests.Session) -> None:
+    def __init__(
+        self,
+        live: Live,
+        session: requests.Session,
+        stream_url_resolver: core_ops.StreamURLResolver,
+    ) -> None:
         self._live = live
         self._session = session
+        self._stream_url_resolver = stream_url_resolver
 
     def __call__(
         self, source: Observable[m3u8.Segment]
     ) -> Observable[Union[InitSectionData, SegmentData]]:
-        return self._fetch(source)
+        return self._fetch(source).pipe(  # type: ignore
+            ops.do_action(on_error=self._before_retry),
+            utils_ops.retry(should_retry=self._should_retry),
+        )
 
     def _fetch(
         self, source: Observable[m3u8.Segment]
@@ -62,10 +81,11 @@ class SegmentFetcher:
             disposed = False
             subscription = SerialDisposable()
 
+            attempts: int = 0
             last_segment: Optional[m3u8.Segment] = None
 
             def on_next(seg: m3u8.Segment) -> None:
-                nonlocal last_segment
+                nonlocal attempts, last_segment
                 url: str = ''
 
                 try:
@@ -115,11 +135,16 @@ class SegmentFetcher:
                             f'segment url: {url}'
                         )
                     else:
-                        raise SegmentDataCorrupted(crc32, crc32_of_data)
+                        logger.warning(f'Segment data corrupted: {url}')
                 except Exception as exc:
                     logger.warning(f'Failed to fetch segment {url}', exc_info=exc)
+                    attempts += 1
+                    if attempts > 3:
+                        attempts = 0
+                        observer.on_error(FetchSegmentError(exc))
                 else:
                     observer.on_next(SegmentData(segment=seg, payload=data))
+                    attempts = 0
 
             def dispose() -> None:
                 nonlocal disposed
@@ -137,8 +162,11 @@ class SegmentFetcher:
 
     @retry(
         reraise=True,
-        retry=retry_if_exception_type(
-            (requests.exceptions.RequestException, urllib3.exceptions.HTTPError)
+        retry=retry_all(
+            retry_if_exception_type(
+                (requests.exceptions.RequestException, urllib3.exceptions.HTTPError)
+            ),
+            retry_if_not_exception_type(requests.exceptions.HTTPError),
         ),
         wait=wait_exponential(max=10),
         stop=stop_after_delay(60),
@@ -147,3 +175,17 @@ class SegmentFetcher:
         with self._session.get(url, headers=self._live.headers, timeout=10) as response:
             response.raise_for_status()
             return response.content
+
+    def _should_retry(self, exc: Exception) -> bool:
+        if isinstance(exc, FetchSegmentError):
+            return True
+        else:
+            return False
+
+    def _before_retry(self, exc: Exception) -> None:
+        if not isinstance(exc, FetchSegmentError):
+            return
+        logger.warning(
+            'Fetch segments failed continuously, trying to update the stream url.'
+        )
+        self._stream_url_resolver.reset()

src/blrec/hls/operators/segment_parser.py

@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+import io
+import logging
+from typing import Optional
+
+from reactivex import Observable, abc
+from reactivex.disposable import CompositeDisposable, Disposable, SerialDisposable
+
+from blrec.flv.common import (
+    is_audio_sequence_header,
+    is_metadata_tag,
+    is_video_sequence_header,
+)
+from blrec.flv.io import FlvReader
+from blrec.flv.models import AudioTag, FlvHeader, ScriptTag, VideoTag
+from blrec.flv.operators.typing import FLVStream, FLVStreamItem
+
+__all__ = ('SegmentParser',)
+
+
+logger = logging.getLogger(__name__)
+
+
+class SegmentParser:
+    def __init__(self) -> None:
+        self._backup_timestamp = True
+
+    def __call__(self, source: Observable[bytes]) -> FLVStream:
+        return self._parse(source)
+
+    def _parse(self, source: Observable[bytes]) -> FLVStream:
+        def subscribe(
+            observer: abc.ObserverBase[FLVStreamItem],
+            scheduler: Optional[abc.SchedulerBase] = None,
+        ) -> abc.DisposableBase:
+            disposed = False
+            subscription = SerialDisposable()
+
+            last_flv_header: Optional[FlvHeader] = None
+            last_metadata_tag: Optional[ScriptTag] = None
+            last_audio_sequence_header: Optional[AudioTag] = None
+            last_video_sequence_header: Optional[VideoTag] = None
+
+            def reset() -> None:
+                nonlocal last_flv_header, last_metadata_tag
+                nonlocal last_audio_sequence_header, last_video_sequence_header
+                last_flv_header = None
+                last_metadata_tag = None
+                last_audio_sequence_header = None
+                last_video_sequence_header = None
+
+            def on_next(data: bytes) -> None:
+                nonlocal last_flv_header, last_metadata_tag
+                nonlocal last_audio_sequence_header, last_video_sequence_header
+
+                if b'' == data:
+                    reset()
+                    return
+
+                try:
+                    reader = FlvReader(
+                        io.BytesIO(data), backup_timestamp=self._backup_timestamp
+                    )
+
+                    flv_header = reader.read_header()
+                    if not last_flv_header:
+                        observer.on_next(flv_header)
+                        last_flv_header = flv_header
+                    else:
+                        assert last_flv_header == flv_header
+
+                    while not disposed:
+                        tag = reader.read_tag()
+                        if is_metadata_tag(tag):
+                            if last_metadata_tag is not None:
+                                continue
+                            last_metadata_tag = tag
+                        elif is_video_sequence_header(tag):
+                            if tag == last_video_sequence_header:
+                                continue
+                            last_video_sequence_header = tag
+                        elif is_audio_sequence_header(tag):
+                            if tag == last_audio_sequence_header:
+                                continue
+                            last_audio_sequence_header = tag
+                        observer.on_next(tag)
+                except EOFError:
+                    pass
+                except Exception as e:
+                    observer.on_error(e)
+
+            def dispose() -> None:
+                nonlocal disposed
+                disposed = True
+                reset()
+
+            subscription.disposable = source.subscribe(
+                on_next, observer.on_error, observer.on_completed, scheduler=scheduler
+            )
+
+            return CompositeDisposable(subscription, Disposable(dispose))
+
+        return Observable(subscribe)

src/blrec/hls/operators/segment_remuxer.py

@@ -2,121 +2,72 @@ from __future__ import annotations
 
 import io
 import logging
-from typing import Final, List, Optional, Union
+import os
+from typing import Optional, Union
 
-import urllib3
+import av
 from reactivex import Observable, abc
 from reactivex.disposable import CompositeDisposable, Disposable, SerialDisposable
-from tenacity import Retrying, stop_after_delay, wait_fixed
-from tenacity.retry import retry_if_not_exception_type
 
 from blrec.bili.live import Live
-from blrec.utils.io import wait_for
 
-from ..stream_remuxer import StreamRemuxer
 from .segment_fetcher import InitSectionData, SegmentData
 
 __all__ = ('SegmentRemuxer',)
 
 
 logger = logging.getLogger(__name__)
-logging.getLogger(urllib3.__name__).setLevel(logging.WARNING)
+
+TRACE_REMUX_SEGMENT = bool(os.environ.get('TRACE_REMUX_SEGMENT'))
+TRACE_LIBAV = bool(os.environ.get('TRACE_LIBAV'))
+if TRACE_LIBAV:
+    logging.getLogger('libav').setLevel(5)
+else:
+    av.logging.set_level(av.logging.FATAL)
 
 
 class SegmentRemuxer:
-    _SEGMENT_DATA_CACHE: Final = 10
-    _MAX_SEGMENT_DATA_CACHE: Final = 15
-
     def __init__(self, live: Live) -> None:
         self._live = live
-        self._timeout: float = 10
-        self._stream_remuxer = StreamRemuxer(live.room_id, remove_filler_data=True)
 
     def __call__(
         self, source: Observable[Union[InitSectionData, SegmentData]]
-    ) -> Observable[io.RawIOBase]:
+    ) -> Observable[bytes]:
         return self._remux(source)
 
     def _remux(
         self, source: Observable[Union[InitSectionData, SegmentData]]
-    ) -> Observable[io.RawIOBase]:
+    ) -> Observable[bytes]:
         def subscribe(
-            observer: abc.ObserverBase[io.RawIOBase],
+            observer: abc.ObserverBase[bytes],
             scheduler: Optional[abc.SchedulerBase] = None,
         ) -> abc.DisposableBase:
             disposed = False
             subscription = SerialDisposable()
 
             init_section_data: Optional[bytes] = None
-            segment_data_cache: List[bytes] = []
-            self._stream_remuxer.stop()
 
             def reset() -> None:
-                nonlocal init_section_data, segment_data_cache
+                nonlocal init_section_data
                 init_section_data = None
-                segment_data_cache = []
-                self._stream_remuxer.stop()
-
-            def write(data: bytes) -> int:
-                return wait_for(
-                    self._stream_remuxer.input.write,
-                    args=(data,),
-                    timeout=self._timeout,
-                )
 
             def on_next(data: Union[InitSectionData, SegmentData]) -> None:
                 nonlocal init_section_data
-                nonlocal segment_data_cache
 
                 if isinstance(data, InitSectionData):
                     init_section_data = data.payload
-                    segment_data_cache.clear()
-                    logger.debug('Stop stream remuxer for init section')
-                    self._stream_remuxer.stop()
+                    observer.on_next(b'')
+                    return
 
-                if self._stream_remuxer.exception and not self._stream_remuxer.stopped:
-                    logger.debug(
-                        'Stop stream remuxer due to '
-                        + repr(self._stream_remuxer.exception)
-                    )
-                    self._stream_remuxer.stop()
+                if init_section_data is None:
+                    return
 
                 try:
-                    if self._stream_remuxer.stopped:
-                        self._stream_remuxer.start()
-                        while True:
-                            ready = self._stream_remuxer.wait(timeout=1)
-                            if disposed:
-                                return
-                            if ready:
-                                break
-
-                        observer.on_next(RemuxedStream(self._stream_remuxer))
-
-                        if init_section_data:
-                            write(init_section_data)
-                        if segment_data_cache:
-                            for cached_data in segment_data_cache:
-                                write(cached_data)
-                        if isinstance(data, InitSectionData):
-                            return
-
-                    write(data.payload)
-                except Exception as e:
-                    logger.warning(f'Failed to write data to stream remuxer: {repr(e)}')
-                    logger.debug(f'Stop stream remuxer due to {repr(e)}')
-                    self._stream_remuxer.stop()
-                    if len(segment_data_cache) >= self._MAX_SEGMENT_DATA_CACHE:
-                        segment_data_cache = segment_data_cache[
-                            -self._MAX_SEGMENT_DATA_CACHE + 1 :
-                        ]
+                    remuxed_data = self._remux_segemnt(init_section_data + data.payload)
+                except av.FFmpegError as e:
+                    logger.warning(f'Failed to remux segment: {repr(e)}', exc_info=e)
                 else:
-                    if len(segment_data_cache) >= self._SEGMENT_DATA_CACHE:
-                        segment_data_cache = segment_data_cache[
-                            -self._SEGMENT_DATA_CACHE + 1 :
-                        ]
-
-                segment_data_cache.append(data.payload)
+                    observer.on_next(remuxed_data)
 
             def dispose() -> None:
                 nonlocal disposed
@@ -131,56 +82,30 @@ class SegmentRemuxer:
 
         return Observable(subscribe)
 
+    def _remux_segemnt(self, data: bytes, format: str = 'flv') -> bytes:
+        in_file = io.BytesIO(data)
+        out_file = io.BytesIO()
 
-class CloseRemuxedStream(Exception):
-    pass
+        with av.open(in_file) as in_container:
+            with av.open(out_file, mode='w', format=format) as out_container:
+                in_video_stream = in_container.streams.video[0]
+                in_audio_stream = in_container.streams.audio[0]
+                out_video_stream = out_container.add_stream(template=in_video_stream)
+                out_audio_stream = out_container.add_stream(template=in_audio_stream)
 
+                for packet in in_container.demux():
+                    if TRACE_REMUX_SEGMENT:
+                        logger.debug(repr(packet))
+                    # We need to skip the "flushing" packets that `demux` generates.
+                    if packet.dts is None:
+                        continue
+                    # We need to assign the packet to the new stream.
+                    if packet.stream.type == 'video':
+                        packet.stream = out_video_stream
+                    elif packet.stream.type == 'audio':
+                        packet.stream = out_audio_stream
+                    else:
+                        raise NotImplementedError(packet.stream.type)
+                    out_container.mux(packet)
 
-class RemuxedStream(io.RawIOBase):
-    def __init__(
-        self, stream_remuxer: StreamRemuxer, *, read_timeout: float = 10
-    ) -> None:
-        self._stream_remuxer = stream_remuxer
-        self._read_timeout = read_timeout
-        self._offset: int = 0
-
-    def read(self, size: int = -1) -> bytes:
-        if self._stream_remuxer.stopped:
-            ready = self._stream_remuxer.wait(timeout=self._read_timeout)
-            if not ready:
-                msg = f'Stream remuxer not ready in {self._read_timeout} seconds'
-                logger.debug(msg)
-                raise EOFError(msg)
-
-        try:
-            for attempt in Retrying(
-                reraise=True,
-                retry=retry_if_not_exception_type(TimeoutError),
-                wait=wait_fixed(1),
-                stop=stop_after_delay(self._read_timeout),
-            ):
-                with attempt:
-                    data = wait_for(
-                        self._stream_remuxer.output.read,
-                        args=(size,),
-                        timeout=self._read_timeout,
-                    )
-        except Exception as exc:
-            logger.warning(f'Failed to read data from stream remuxer: {repr(exc)}')
-            self._stream_remuxer.exception = exc
-            raise EOFError(exc)
-        else:
-            assert data is not None
-            self._offset += len(data)
-            return data
-
-    def tell(self) -> int:
-        return self._offset
-
-    def close(self) -> None:
-        if self._stream_remuxer.stopped:
-            return
-        if self._stream_remuxer.exception:
-            return
-        logger.debug('Close remuxed stream')
-        self._stream_remuxer.exception = CloseRemuxedStream()
+        return out_file.getvalue()

src/blrec/hls/stream_remuxer.py

@@ -1,164 +0,0 @@
-import errno
-import io
-import logging
-import os
-import re
-import shlex
-from contextlib import suppress
-from subprocess import PIPE, CalledProcessError, Popen
-from threading import Condition, Thread
-from typing import Optional, cast
-
-from blrec.utils.io import wait_for
-from blrec.utils.mixins import StoppableMixin, SupportDebugMixin
-
-logger = logging.getLogger(__name__)
-
-
-__all__ = ('StreamRemuxer',)
-
-
-class FFmpegError(Exception):
-    pass
-
-
-class StreamRemuxer(StoppableMixin, SupportDebugMixin):
-    _ERROR_PATTERN = re.compile(
-        r'\b(error|failed|missing|invalid|corrupt)\b', re.IGNORECASE
-    )
-
-    def __init__(self, room_id: int, remove_filler_data: bool = False) -> None:
-        super().__init__()
-        self._room_id = room_id
-        self._remove_filler_data = remove_filler_data
-        self._exception: Optional[Exception] = None
-        self._ready = Condition()
-        self._env = None
-
-        self._init_for_debug(room_id)
-        if self._debug:
-            self._env = os.environ.copy()
-            path = os.path.join(self._debug_dir, f'ffreport-{room_id}-%t.log')
-            self._env['FFREPORT'] = f'file={path}:level=48'
-
-    @property
-    def input(self) -> io.BufferedWriter:
-        assert self._subprocess.stdin is not None
-        return cast(io.BufferedWriter, self._subprocess.stdin)
-
-    @property
-    def output(self) -> io.BufferedReader:
-        assert self._subprocess.stdout is not None
-        return cast(io.BufferedReader, self._subprocess.stdout)
-
-    @property
-    def exception(self) -> Optional[Exception]:
-        return self._exception
-
-    @exception.setter
-    def exception(self, exc: Exception) -> None:
-        self._exception = exc
-
-    def __enter__(self):  # type: ignore
-        self.start()
-        self.wait()
-        return self
-
-    def __exit__(self, exc_type, value, traceback):  # type: ignore
-        self.stop()
-        self.raise_for_exception()
-
-    def wait(self, timeout: Optional[float] = None) -> bool:
-        with self._ready:
-            return self._ready.wait(timeout=timeout)
-
-    def restart(self) -> None:
-        logger.debug('Restarting stream remuxer...')
-        self.stop()
-        self.start()
-        logger.debug('Restarted stream remuxer')
-
-    def raise_for_exception(self) -> None:
-        if not self.exception:
-            return
-        raise self.exception
-
-    def _do_start(self) -> None:
-        logger.debug('Starting stream remuxer...')
-        self._thread = Thread(
-            target=self._run, name=f'StreamRemuxer::{self._room_id}', daemon=True
-        )
-        self._thread.start()
-
-    def _do_stop(self) -> None:
-        logger.debug('Stopping stream remuxer...')
-        if hasattr(self, '_subprocess'):
-            with suppress(ProcessLookupError):
-                self._subprocess.kill()
-            self._subprocess.wait(timeout=10)
-        if hasattr(self, '_thread'):
-            self._thread.join(timeout=10)
-
-    def _run(self) -> None:
-        logger.debug('Started stream remuxer')
-        self._exception = None
-        try:
-            self._run_subprocess()
-        except BrokenPipeError as exc:
-            logger.debug(repr(exc))
-        except FFmpegError as exc:
-            if not self._stopped:
-                logger.warning(repr(exc))
-            else:
-                logger.debug(repr(exc))
-        except TimeoutError as exc:
-            logger.debug(repr(exc))
-        except Exception as exc:
-            # OSError: [Errno 22] Invalid argument
-            # https://stackoverflow.com/questions/23688492/oserror-errno-22-invalid-argument-in-subprocess
-            if isinstance(exc, OSError) and exc.errno == errno.EINVAL:
-                pass
-            else:
-                self._exception = exc
-                logger.exception(exc)
-        finally:
-            self._stopped = True
-            logger.debug('Stopped stream remuxer')
-
-    def _run_subprocess(self) -> None:
-        cmd = 'ffmpeg -xerror -i pipe:0 -c copy -copyts'
-        if self._remove_filler_data:
-            cmd += ' -bsf:v filter_units=remove_types=12'
-        cmd += ' -f flv pipe:1'
-        args = shlex.split(cmd)
-
-        with Popen(
-            args, stdin=PIPE, stdout=PIPE, stderr=PIPE, env=self._env
-        ) as self._subprocess:
-            with self._ready:
-                self._ready.notify_all()
-
-            assert self._subprocess.stderr is not None
-            with io.TextIOWrapper(
-                self._subprocess.stderr, encoding='utf-8', errors='backslashreplace'
-            ) as stderr:
-                while not self._stopped:
-                    line = wait_for(stderr.readline, timeout=10)
-                    if not line:
-                        if self._subprocess.poll() is not None:
-                            break
-                        else:
-                            continue
-                    if self._debug:
-                        logger.debug('ffmpeg: %s', line)
-                    self._check_error(line)
-
-        if not self._stopped and self._subprocess.returncode not in (0, 255):
-            # 255: Exiting standardly, received signal 2.
-            raise CalledProcessError(self._subprocess.returncode, cmd=cmd)
-
-    def _check_error(self, line: str) -> None:
-        match = self._ERROR_PATTERN.search(line)
-        if not match:
-            return
-        raise FFmpegError(line)

src/blrec/logging/configure_logging.py

@@ -1,19 +1,18 @@
-import os
-import logging
-from logging import LogRecord, Handler
-from logging.handlers import RotatingFileHandler
-from datetime import datetime
 import asyncio
-import threading
 import atexit
+import logging
+import os
+import threading
+from datetime import datetime
+from logging import Handler, LogRecord
+from logging.handlers import TimedRotatingFileHandler
 from typing import Any, List, Optional
 
-from colorama import init, deinit, Fore, Back, Style
+from colorama import Back, Fore, Style, deinit, init
 from tqdm import tqdm
 
 from .typing import LOG_LEVEL
 
-
 __all__ = 'configure_logger', 'ConsoleHandler', 'TqdmOutputStream'
 
 
@@ -57,7 +56,7 @@ def obtain_room_id() -> str:
         name = task.get_name()
 
     if '::' in name:
-        if (room_id := name.split('::')[-1]):
+        if room_id := name.split('::')[-1]:
             return room_id
 
     return ''
@@ -66,7 +65,7 @@ def obtain_room_id() -> str:
 def record_factory(*args: Any, **kwargs: Any) -> LogRecord:
     record = _old_factory(*args, **kwargs)
 
-    if (room_id := obtain_room_id()):
+    if room_id := obtain_room_id():
         record.roomid = '[' + room_id + '] '  # type: ignore
     else:
         record.roomid = ''  # type: ignore
@@ -84,7 +83,6 @@ def configure_logger(
     log_dir: str,
     *,
     console_log_level: LOG_LEVEL = 'INFO',
-    max_bytes: Optional[int] = None,
     backup_count: Optional[int] = None,
 ) -> None:
     # config root logger
@@ -104,10 +102,10 @@ def configure_logger(
 
     # logging to file
     log_file_path = make_log_file_path(log_dir)
-    file_handler = RotatingFileHandler(
-        log_file_path,
-        maxBytes=max_bytes or 1024 ** 2 * 10,
-        backupCount=backup_count or 1,
+    file_handler = TimedRotatingFileHandler(
+        filename=log_file_path,
+        when='MIDNIGHT',
+        backupCount=backup_count or 0,
         encoding='utf-8',
     )
     file_handler.setLevel(logging.DEBUG)

src/blrec/notification/__init__.py

@@ -6,6 +6,7 @@ from .notifiers import (
     PushdeerNotifier,
     PushplusNotifier,
     TelegramNotifier,
+    BarkNotifier,
 )
 from .providers import (
     MessagingProvider,
@@ -14,6 +15,7 @@ from .providers import (
     Pushdeer,
     Pushplus,
     Telegram,
+    Bark,
 )
 
 
@@ -24,6 +26,7 @@ __all__ = (
     'Pushdeer',
     'Pushplus',
     'Telegram',
+    'Bark',
 
     'Notifier',
     'MessageNotifier',
@@ -32,4 +35,5 @@ __all__ = (
     'PushdeerNotifier',
     'PushplusNotifier',
     'TelegramNotifier',
+    'BarkNotifier',
 )

src/blrec/notification/notifiers.py

@@ -36,6 +36,7 @@ from .providers import (
     Pushplus,
     Serverchan,
     Telegram,
+    Bark,
 )
 
 __all__ = (
@@ -46,6 +47,7 @@ __all__ = (
     'PushdeerNotifier',
     'PushplusNotifier',
     'TelegramNotifier',
+    'BarkNotifer',
 )
 
 
@@ -403,3 +405,15 @@ class TelegramNotifier(MessageNotifier):
 
     def _get_error_message_content(self, msg_type: Optional[MessageType] = None) -> str:
         return super()._get_error_message_content(msg_type='text')
+
+
+class BarkNotifier(MessageNotifier):
+    provider = Bark.get_instance()
+
+    def _do_enable(self) -> None:
+        super()._do_enable()
+        logger.debug('Enabled Bark notifier')
+
+    def _do_disable(self) -> None:
+        super()._do_disable()
+        logger.debug('Disabled Bark notifier')

src/blrec/notification/providers.py

@@ -17,6 +17,7 @@ from ..setting.typing import (
     PushplusMessageType,
     ServerchanMessageType,
     TelegramMessageType,
+    BarkMessageType,
 )
 from ..utils.patterns import Singleton
 
@@ -27,6 +28,7 @@ __all__ = (
     'Pushdeer',
     'Pushplus',
     'Telegram',
+    'Bark',
 )
 
 
@@ -251,3 +253,50 @@ class Telegram(MessagingProvider):
                         response['result']['error_code'],
                         response['result']['description'],
                     )
+
+
+class BarkResponse(TypedDict):
+    code: int
+    message: str
+    timestamp: int
+
+
+class Bark(MessagingProvider):
+    _server: Final = 'https://api.day.app'
+    _endpoint: Final = '/push'
+
+    def __init__(self, server: str = '', pushkey: str = '') -> None:
+        super().__init__()
+        self.server = server
+        self.pushkey = pushkey
+
+    async def send_message(
+        self, title: str, content: str, msg_type: MessageType
+    ) -> None:
+        self._check_parameters()
+        await self._post_message(title, content, cast(BarkMessageType, msg_type))
+
+    def _check_parameters(self) -> None:
+        if not self.pushkey:
+            raise ValueError('No pushkey supplied')
+
+    async def _post_message(
+        self, title: str, content: str, msg_type: BarkMessageType
+    ) -> None:
+        url = urljoin(self.server or self._server, self._endpoint)
+        # content size is limited to a maximum size of 4 KB (4096 bytes)
+        if len(content.encode()) >= 4096:
+            content = content.encode()[:4090].decode(errors='ignore') + ' ...'
+        payload = {
+            "title": title,
+            "body": content,
+            "device_key": self.pushkey,
+            "badge": 1,
+            "icon": "https://raw.githubusercontent.com/acgnhiki/blrec/master/webapp/src/assets/icons/icon-72x72.png",
+            "group": "blrec",
+        }
+        async with aiohttp.ClientSession(raise_for_status=True) as session:
+            async with session.post(url, json=payload) as res:
+                response = cast(BarkResponse, await res.json())
+                if response['code'] != 200:
+                    raise HTTPException(response['message'])

src/blrec/postprocess/postprocessor.py

@@ -176,7 +176,10 @@ class Postprocessor(
                             await copy_files_related(video_path)
                             if result_path != video_path:
                                 self._completed_files.append(danmaku_path(result_path))
-                                self._completed_files.remove(danmaku_path(video_path))
+                                with suppress(ValueError):
+                                    self._completed_files.remove(
+                                        danmaku_path(video_path)
+                                    )
                             if not self._debug:
                                 if self._should_delete_source_files(remuxing_result):
                                     await discard_dir(os.path.dirname(video_path))

src/blrec/setting/__init__.py

@@ -36,6 +36,9 @@ from .models import (
     TelegramMessageTemplateSettings,
     TelegramNotificationSettings,
     TelegramSettings,
+    BarkMessageTemplateSettings,
+    BarkNotificationSettings,
+    BarkSettings,
     WebHookSettings,
 )
 from .setting_manager import SettingsManager
@@ -65,11 +68,13 @@ __all__ = (
     'PushdeerMessageTemplateSettings',
     'PushplusMessageTemplateSettings',
     'TelegramMessageTemplateSettings',
+    'BarkMessageTemplateSettings',
     'EmailSettings',
     'ServerchanSettings',
     'PushdeerSettings',
     'PushplusSettings',
     'TelegramSettings',
+    'BarkSettings',
     'NotifierSettings',
     'NotificationSettings',
     'EmailNotificationSettings',
@@ -77,6 +82,7 @@ __all__ = (
     'PushdeerNotificationSettings',
     'PushplusNotificationSettings',
     'TelegramNotificationSettings',
+    'BarkNotificationSettings',
     'WebHookSettings',
     'update_settings',
     'shadow_settings',

src/blrec/setting/models.py

@@ -24,6 +24,7 @@ from .typing import (
     RecordingMode,
     ServerchanMessageType,
     TelegramMessageType,
+    BarkMessageType,
 )
 
 logger = logging.getLogger(__name__)
@@ -54,6 +55,7 @@ __all__ = (
     'PushdeerSettings',
     'PushplusSettings',
     'TelegramSettings',
+    'BarkSettings'
     'NotifierSettings',
     'NotificationSettings',
     'EmailMessageTemplateSettings',
@@ -61,11 +63,13 @@ __all__ = (
     'PushdeerMessageTemplateSettings',
     'PushplusMessageTemplateSettings',
     'TelegramMessageTemplateSettings',
+    'BarkMessageTemplateSettings',
     'EmailNotificationSettings',
     'ServerchanNotificationSettings',
     'PushdeerNotificationSettings',
     'PushplusNotificationSettings',
     'TelegramNotificationSettings',
+    'BarkNotificationSettings',
     'WebHookSettings',
 )
 
@@ -322,12 +326,7 @@ def log_dir_factory() -> str:
 class LoggingSettings(BaseModel):
     log_dir: Annotated[str, Field(default_factory=log_dir_factory)]
     console_log_level: LOG_LEVEL = 'INFO'
-    max_bytes: Annotated[
-        int, Field(ge=1024**2, le=1024**2 * 10, multiple_of=1024**2)
-    ] = (
-        1024**2 * 10
-    )  # allowed 1 ~ 10 MB
-    backup_count: Annotated[int, Field(ge=1, le=30)] = 30
+    backup_count: Annotated[int, Field(ge=0, le=90)] = 30
 
     @validator('log_dir')
     def _validate_dir(cls, path: str) -> str:
@@ -419,6 +418,23 @@ class TelegramSettings(BaseModel):
         return value
 
 
+class BarkSettings(BaseModel):
+    server: str = ''
+    pushkey: str = ''
+
+    @validator('server')
+    def _validate_server(cls, value: str) -> str:
+        if value != '' and not re.fullmatch(r'https?://.+', value):
+            raise ValueError('server is invalid')
+        return value
+
+    @validator('pushkey')
+    def _validate_pushkey(cls, value: str) -> str:
+        if value != '' and not re.fullmatch(r'[a-zA-Z\d]+', value):
+            raise ValueError('pushkey is invalid')
+        return value
+
+
 class NotifierSettings(BaseModel):
     enabled: bool = False
 
@@ -520,6 +536,21 @@ class TelegramMessageTemplateSettings(MessageTemplateSettings):
     error_message_content: str = ''
 
 
+class BarkMessageTemplateSettings(MessageTemplateSettings):
+    began_message_type: BarkMessageType = 'markdown'
+    began_message_title: str = ''
+    began_message_content: str = ''
+    ended_message_type: BarkMessageType = 'markdown'
+    ended_message_title: str = ''
+    ended_message_content: str = ''
+    space_message_type: BarkMessageType = 'markdown'
+    space_message_title: str = ''
+    space_message_content: str = ''
+    error_message_type: BarkMessageType = 'markdown'
+    error_message_title: str = ''
+    error_message_content: str = ''
+
+
 class EmailNotificationSettings(
     EmailSettings, NotifierSettings, NotificationSettings, EmailMessageTemplateSettings
 ):
@@ -562,6 +593,15 @@ class TelegramNotificationSettings(
     pass
 
 
+class BarkNotificationSettings(
+    BarkSettings,
+    NotifierSettings,
+    NotificationSettings,
+    BarkMessageTemplateSettings,
+):
+    pass
+
+
 class WebHookEventSettings(BaseModel):
     live_began: bool = True
     live_ended: bool = True
@@ -607,6 +647,7 @@ class Settings(BaseModel):
     pushdeer_notification: PushdeerNotificationSettings = PushdeerNotificationSettings()
     pushplus_notification: PushplusNotificationSettings = PushplusNotificationSettings()
     telegram_notification: TelegramNotificationSettings = TelegramNotificationSettings()
+    bark_notification: BarkNotificationSettings = BarkNotificationSettings()
     webhooks: Annotated[List[WebHookSettings], Field(max_items=50)] = []
 
     @classmethod
@@ -655,6 +696,7 @@ class SettingsIn(BaseModel):
     pushdeer_notification: Optional[PushdeerNotificationSettings] = None
     pushplus_notification: Optional[PushplusNotificationSettings] = None
     telegram_notification: Optional[TelegramNotificationSettings] = None
+    bark_notification: Optional[BarkNotificationSettings] = None
     webhooks: Optional[List[WebHookSettings]] = None
 
 

src/blrec/setting/setting_manager.py

@@ -12,6 +12,7 @@ from ..notification import (
     Pushplus,
     Serverchan,
     Telegram,
+    Bark,
 )
 from ..webhook import WebHook
 from .helpers import shadow_settings, update_settings
@@ -264,7 +265,6 @@ class SettingsManager:
         configure_logger(
             log_dir=self._settings.logging.log_dir,
             console_log_level=self._settings.logging.console_log_level,
-            max_bytes=self._settings.logging.max_bytes,
             backup_count=self._settings.logging.backup_count,
         )
 
@@ -345,6 +345,14 @@ class SettingsManager:
         self._apply_notification_settings(notifier, settings)
         self._apply_message_template_settings(notifier, settings)
 
+    def apply_bark_notification_settings(self) -> None:
+        notifier = self._app._bark_notifier
+        settings = self._settings.bark_notification
+        self._apply_bark_settings(notifier.provider)
+        self._apply_notifier_settings(notifier, settings)
+        self._apply_notification_settings(notifier, settings)
+        self._apply_message_template_settings(notifier, settings)
+
     def apply_webhooks_settings(self) -> None:
         webhooks = [WebHook.from_settings(s) for s in self._settings.webhooks]
         self._app._webhook_emitter.webhooks = webhooks
@@ -371,6 +379,10 @@ class SettingsManager:
         telegram.token = self._settings.telegram_notification.token
         telegram.chatid = self._settings.telegram_notification.chatid
 
+    def _apply_bark_settings(self, bark: Bark) -> None:
+        bark.server = self._settings.bark_notification.server
+        bark.pushkey = self._settings.bark_notification.pushkey
+
     def _apply_notifier_settings(
         self, notifier: Notifier, settings: NotifierSettings
     ) -> None:

src/blrec/setting/typing.py

@@ -12,6 +12,7 @@ ServerchanMessageType = MarkdownMessageType
 PushdeerMessageType = Union[TextMessageType, MarkdownMessageType]
 PushplusMessageType = Union[TextMessageType, MarkdownMessageType, HtmlMessageType]
 TelegramMessageType = Union[MarkdownMessageType, HtmlMessageType]
+BarkMessageType = Union[TextMessageType, MarkdownMessageType]
 
 
 KeyOfSettings = Literal[
@@ -30,6 +31,7 @@ KeyOfSettings = Literal[
     'pushdeer_notification',
     'pushplus_notification',
     'telegram_notification',
+    'bark_notification',
     'webhooks',
 ]
 

src/blrec/web/schemas.py

@@ -55,5 +55,6 @@ AliasKeyOfSettings = Literal[
     'pushdeerNotification',
     'pushplusNotification',
     'telegramNotification',
+    'barkNotification',
     'webhooks',
 ]

webapp/src/app/settings/logging-settings/logging-settings.component.html

@@ -36,19 +36,6 @@
       </nz-select>
     </nz-form-control>
   </nz-form-item>
-  <nz-form-item class="setting-item">
-    <nz-form-label class="setting-label" nzNoColon
-      >日志文件分割大小</nz-form-label
-    >
-    <nz-form-control
-      class="setting-control select"
-      [nzWarningTip]="syncFailedWarningTip"
-      [nzValidateStatus]="syncStatus.maxBytes ? maxBytesControl : 'warning'"
-    >
-      <nz-select formControlName="maxBytes" [nzOptions]="maxBytesOptions">
-      </nz-select>
-    </nz-form-control>
-  </nz-form-item>
   <nz-form-item class="setting-item">
     <nz-form-label class="setting-label" nzNoColon
       >日志文件备份数量</nz-form-label

webapp/src/app/settings/logging-settings/logging-settings.component.ts

@@ -42,12 +42,7 @@ export class LoggingSettingsComponent implements OnInit, OnChanges {
     { label: 'CRITICAL', value: 'CRITICAL' },
   ];
 
-  readonly maxBytesOptions = range(1, 11).map((i) => ({
-    label: `${i} MB`,
-    value: 1024 ** 2 * i,
-  }));
-
-  readonly backupOptions = range(1, 31).map((i) => ({
+  readonly backupOptions = range(0, 91).map((i) => ({
     label: i.toString(),
     value: i,
   }));
@@ -60,7 +55,6 @@ export class LoggingSettingsComponent implements OnInit, OnChanges {
     this.settingsForm = formBuilder.group({
       logDir: [''],
       consoleLogLevel: [''],
-      maxBytes: [''],
       backupCount: [''],
     });
   }
@@ -73,10 +67,6 @@ export class LoggingSettingsComponent implements OnInit, OnChanges {
     return this.settingsForm.get('consoleLogLevel') as FormControl;
   }
 
-  get maxBytesControl() {
-    return this.settingsForm.get('maxBytes') as FormControl;
-  }
-
   get backupCountControl() {
     return this.settingsForm.get('backupCount') as FormControl;
   }

webapp/src/app/settings/notification-settings/bark-notification-settings/bark-notification-settings.component.html

@@ -0,0 +1,20 @@
+<app-sub-page pageTitle="Bark 通知">
+  <ng-template appSubPageContent>
+    <app-page-section>
+      <app-notifier-settings [settings]="notifierSettings" keyOfSettings="barkNotification"></app-notifier-settings>
+    </app-page-section>
+
+    <app-page-section name="Bark">
+      <app-bark-settings [settings]="barkSettings"></app-bark-settings>
+    </app-page-section>
+
+    <app-page-section name="事件">
+      <app-event-settings [settings]="notificationSettings" keyOfSettings="barkNotification"></app-event-settings>
+    </app-page-section>
+
+    <app-page-section name="消息">
+      <app-message-template-settings [settings]="messageTemplateSettings" keyOfSettings="barkNotification">
+      </app-message-template-settings>
+    </app-page-section>
+  </ng-template>
+</app-sub-page>

webapp/src/app/settings/notification-settings/bark-notification-settings/bark-notification-settings.component.scss

@@ -0,0 +1,2 @@
+
+@use '../../shared/styles/setting';

webapp/src/app/settings/notification-settings/bark-notification-settings/bark-notification-settings.component.spec.ts

@@ -0,0 +1,25 @@
+import { ComponentFixture, TestBed } from '@angular/core/testing';
+
+import { BarkNotificationSettingsComponent } from './bark-notification-settings.component';
+
+describe('BarkNotificationSettingsComponent', () => {
+  let component: BarkNotificationSettingsComponent;
+  let fixture: ComponentFixture<BarkNotificationSettingsComponent>;
+
+  beforeEach(async () => {
+    await TestBed.configureTestingModule({
+      declarations: [BarkNotificationSettingsComponent]
+    })
+      .compileComponents();
+  });
+
+  beforeEach(() => {
+    fixture = TestBed.createComponent(BarkNotificationSettingsComponent);
+    component = fixture.componentInstance;
+    fixture.detectChanges();
+  });
+
+  it('should create', () => {
+    expect(component).toBeTruthy();
+  });
+});

webapp/src/app/settings/notification-settings/bark-notification-settings/bark-notification-settings.component.ts

@@ -0,0 +1,53 @@
+import {
+  Component,
+  OnInit,
+  ChangeDetectionStrategy,
+  ChangeDetectorRef,
+} from '@angular/core';
+import { ActivatedRoute } from '@angular/router';
+
+import pick from 'lodash-es/pick';
+
+import {
+  KEYS_OF_MESSAGE_TEMPLATE_SETTINGS,
+  KEYS_OF_NOTIFICATION_SETTINGS,
+  KEYS_OF_NOTIFIER_SETTINGS,
+  KEYS_OF_BARK_SETTINGS,
+  MessageTemplateSettings,
+  NotificationSettings,
+  NotifierSettings,
+  BarkNotificationSettings,
+  BarkSettings,
+} from '../../shared/setting.model';
+
+@Component({
+  selector: 'app-bark-notification-settings',
+  templateUrl: './bark-notification-settings.component.html',
+  styleUrls: ['./bark-notification-settings.component.scss'],
+  changeDetection: ChangeDetectionStrategy.OnPush,
+})
+export class BarkNotificationSettingsComponent implements OnInit {
+  barkSettings!: BarkSettings;
+  notifierSettings!: NotifierSettings;
+  notificationSettings!: NotificationSettings;
+  messageTemplateSettings!: MessageTemplateSettings;
+
+  constructor(
+    private changeDetector: ChangeDetectorRef,
+    private route: ActivatedRoute
+  ) { }
+
+  ngOnInit(): void {
+    this.route.data.subscribe((data) => {
+      const settings = data.settings as BarkNotificationSettings;
+      this.barkSettings = pick(settings, KEYS_OF_BARK_SETTINGS);
+      this.notifierSettings = pick(settings, KEYS_OF_NOTIFIER_SETTINGS);
+      this.notificationSettings = pick(settings, KEYS_OF_NOTIFICATION_SETTINGS);
+      this.messageTemplateSettings = pick(
+        settings,
+        KEYS_OF_MESSAGE_TEMPLATE_SETTINGS
+      );
+      this.changeDetector.markForCheck();
+    });
+  }
+}

webapp/src/app/settings/notification-settings/bark-notification-settings/bark-settings/bark-settings.component.html

@@ -0,0 +1,33 @@
+<form nz-form [formGroup]="settingsForm">
+  <nz-form-item class="setting-item">
+    <nz-form-label class="setting-label align-required" nzFor="server" nzNoColon>server</nz-form-label>
+    <nz-form-control class="setting-control input" nzHasFeedback [nzErrorTip]="serverErrorTip"
+      [nzWarningTip]="syncFailedWarningTip" [nzValidateStatus]="
+        serverControl.valid && !syncStatus.server ? 'warning' : serverControl
+      ">
+      <input id="server" type="url" placeholder="默认为官方服务器 https://api.day.app" nz-input formControlName="server" />
+      <ng-template #serverErrorTip let-control>
+        <ng-container *ngIf="control.hasError('pattern')">
+          server 无效
+        </ng-container>
+      </ng-template>
+    </nz-form-control>
+  </nz-form-item>
+  <nz-form-item class="setting-item">
+    <nz-form-label class="setting-label" nzFor="pushkey" nzNoColon nzRequired>pushkey</nz-form-label>
+    <nz-form-control class="setting-control input" nzHasFeedback [nzErrorTip]="pushkeyErrorTip"
+      [nzWarningTip]="syncFailedWarningTip" [nzValidateStatus]="
+        pushkeyControl.valid && !syncStatus.pushkey ? 'warning' : pushkeyControl
+      ">
+      <input id="pushkey" type="text" placeholder="" required nz-input formControlName="pushkey" />
+      <ng-template #pushkeyErrorTip let-control>
+        <ng-container *ngIf="control.hasError('required')">
+          请输入 pushkey！
+        </ng-container>
+        <ng-container *ngIf="control.hasError('pattern')">
+          pushkey 无效
+        </ng-container>
+      </ng-template>
+    </nz-form-control>
+  </nz-form-item>
+</form>

webapp/src/app/settings/notification-settings/bark-notification-settings/bark-settings/bark-settings.component.scss

@@ -0,0 +1,6 @@
+@use '../../../shared/styles/setting';
+
+.setting-label {
+  max-width: 5em !important;
+  width: 5em !important;
+}

webapp/src/app/settings/notification-settings/bark-notification-settings/bark-settings/bark-settings.component.spec.ts

@@ -0,0 +1,25 @@
+import { ComponentFixture, TestBed } from '@angular/core/testing';
+
+import { BarkSettingsComponent } from './bark-settings.component';
+
+describe('BarkSettingsComponent', () => {
+  let component: BarkSettingsComponent;
+  let fixture: ComponentFixture<BarkSettingsComponent>;
+
+  beforeEach(async () => {
+    await TestBed.configureTestingModule({
+      declarations: [BarkSettingsComponent]
+    })
+      .compileComponents();
+  });
+
+  beforeEach(() => {
+    fixture = TestBed.createComponent(BarkSettingsComponent);
+    component = fixture.componentInstance;
+    fixture.detectChanges();
+  });
+
+  it('should create', () => {
+    expect(component).toBeTruthy();
+  });
+});

webapp/src/app/settings/notification-settings/bark-notification-settings/bark-settings/bark-settings.component.ts

@@ -0,0 +1,87 @@
+import {
+  Component,
+  OnInit,
+  ChangeDetectionStrategy,
+  Input,
+  OnChanges,
+  ChangeDetectorRef,
+} from '@angular/core';
+import {
+  FormBuilder,
+  FormControl,
+  FormGroup,
+  Validators,
+} from '@angular/forms';
+
+import mapValues from 'lodash-es/mapValues';
+
+import { BarkSettings } from '../../../shared/setting.model';
+import { filterValueChanges } from '../../../shared/rx-operators';
+import {
+  SettingsSyncService,
+  SyncStatus,
+  calcSyncStatus,
+} from '../../../shared/services/settings-sync.service';
+import { SYNC_FAILED_WARNING_TIP } from 'src/app/settings/shared/constants/form';
+
+@Component({
+  selector: 'app-bark-settings',
+  templateUrl: './bark-settings.component.html',
+  styleUrls: ['./bark-settings.component.scss'],
+  changeDetection: ChangeDetectionStrategy.OnPush,
+})
+export class BarkSettingsComponent implements OnInit, OnChanges {
+  @Input() settings!: BarkSettings;
+  syncStatus!: SyncStatus<BarkSettings>;
+
+  readonly settingsForm: FormGroup;
+  readonly syncFailedWarningTip = SYNC_FAILED_WARNING_TIP;
+
+  constructor(
+    formBuilder: FormBuilder,
+    private changeDetector: ChangeDetectorRef,
+    private settingsSyncService: SettingsSyncService
+  ) {
+    this.settingsForm = formBuilder.group({
+      server: ['', [Validators.pattern(/^https?:\/\/.+/)]],
+      pushkey: [
+        '',
+        [
+          Validators.required,
+          Validators.pattern(
+            /^[a-zA-Z\d]+$/
+          ),
+        ],
+      ],
+    });
+  }
+
+  get serverControl() {
+    return this.settingsForm.get('server') as FormControl;
+  }
+
+  get pushkeyControl() {
+    return this.settingsForm.get('pushkey') as FormControl;
+  }
+
+  ngOnChanges(): void {
+    this.syncStatus = mapValues(this.settings, () => true);
+    console.log(this.settings);
+    this.settingsForm.setValue(this.settings);
+  }
+
+  ngOnInit(): void {
+    this.settingsSyncService
+      .syncSettings(
+        'barkNotification',
+        this.settings,
+        this.settingsForm.valueChanges.pipe(
+          filterValueChanges<Partial<BarkSettings>>(this.settingsForm)
+        )
+      )
+      .subscribe((detail) => {
+        this.syncStatus = { ...this.syncStatus, ...calcSyncStatus(detail) };
+        this.changeDetector.markForCheck();
+      });
+  }
+}

webapp/src/app/settings/notification-settings/notification-settings.component.html

@@ -1,20 +1,12 @@
-<a class="setting-item" routerLink="email-notification"
-  ><span class="setting-label">邮箱通知</span
-  ><span class="setting-control"> <i nz-icon nzType="right"></i> </span
-></a>
-<a class="setting-item" routerLink="serverchan-notification"
-  ><span class="setting-label">ServerChan 通知</span
-  ><span class="setting-control"><i nz-icon nzType="right"></i></span
-></a>
-<a class="setting-item" routerLink="pushdeer-notification"
-  ><span class="setting-label">PushDeer 通知</span
-  ><span class="setting-control"><i nz-icon nzType="right"></i></span
-></a>
-<a class="setting-item" routerLink="pushplus-notification"
-  ><span class="setting-label">pushplus 通知</span
-  ><span class="setting-control"><i nz-icon nzType="right"></i></span
-></a>
-<a class="setting-item" routerLink="telegram-notification"
-  ><span class="setting-label">telegram 通知</span
-  ><span class="setting-control"><i nz-icon nzType="right"></i></span
-></a>
+<a class="setting-item" routerLink="email-notification"><span class="setting-label">邮箱通知</span><span
+    class="setting-control"> <i nz-icon nzType="right"></i> </span></a>
+<a class="setting-item" routerLink="serverchan-notification"><span class="setting-label">ServerChan 通知</span><span
+    class="setting-control"><i nz-icon nzType="right"></i></span></a>
+<a class="setting-item" routerLink="pushdeer-notification"><span class="setting-label">PushDeer 通知</span><span
+    class="setting-control"><i nz-icon nzType="right"></i></span></a>
+<a class="setting-item" routerLink="pushplus-notification"><span class="setting-label">pushplus 通知</span><span
+    class="setting-control"><i nz-icon nzType="right"></i></span></a>
+<a class="setting-item" routerLink="telegram-notification"><span class="setting-label">telegram 通知</span><span
+    class="setting-control"><i nz-icon nzType="right"></i></span></a>
+<a class="setting-item" routerLink="bark-notification"><span class="setting-label">Bark 通知</span><span
+    class="setting-control"><i nz-icon nzType="right"></i></span></a>

webapp/src/app/settings/notification-settings/pushdeer-notification-settings/pushdeer-settings/pushdeer-settings.component.ts

@@ -66,6 +66,7 @@ export class PushdeerSettingsComponent implements OnInit, OnChanges {
 
   ngOnChanges(): void {
     this.syncStatus = mapValues(this.settings, () => true);
+    console.log(this.settings);
     this.settingsForm.setValue(this.settings);
   }
 

webapp/src/app/settings/notification-settings/shared/components/event-settings/event-settings.component.ts

@@ -32,7 +32,8 @@ export class EventSettingsComponent implements OnInit, OnChanges {
     | 'serverchanNotification'
     | 'pushdeerNotification'
     | 'pushplusNotification'
-    | 'telegramNotification';
+    | 'telegramNotification'
+    | 'barkNotification';
 
   syncStatus!: SyncStatus<NotificationSettings>;
 

webapp/src/app/settings/notification-settings/shared/components/message-template-settings/message-template-settings.component.ts

@@ -38,7 +38,8 @@ export class MessageTemplateSettingsComponent implements OnInit, OnChanges {
     | 'serverchanNotification'
     | 'pushdeerNotification'
     | 'pushplusNotification'
-    | 'telegramNotification';
+    | 'telegramNotification'
+    | 'barkNotification';
 
   messageTypes!: MessageType[];
   beganMessageTemplateSettings!: CommonMessageTemplateSettings;
@@ -50,7 +51,7 @@ export class MessageTemplateSettingsComponent implements OnInit, OnChanges {
     private changeDetector: ChangeDetectorRef,
     private message: NzMessageService,
     private settingService: SettingService
-  ) {}
+  ) { }
 
   ngOnInit(): void {
     switch (this.keyOfSettings) {

webapp/src/app/settings/notification-settings/shared/components/notifier-settings/notifier-settings.component.ts

@@ -32,7 +32,8 @@ export class NotifierSettingsComponent implements OnInit, OnChanges {
     | 'serverchanNotification'
     | 'pushdeerNotification'
     | 'pushplusNotification'
-    | 'telegramNotification';
+    | 'telegramNotification'
+    | 'barkNotification';
 
   syncStatus!: SyncStatus<NotifierSettings>;
 

webapp/src/app/settings/settings-routing.module.ts

@@ -7,6 +7,7 @@ import { PushplusNotificationSettingsResolver } from './shared/services/pushplus
 import { TelegramNotificationSettingsResolver } from './shared/services/telegram-notification-settings.resolver';
 import { ServerchanNotificationSettingsResolver } from './shared/services/serverchan-notification-settings.resolver';
 import { PushdeerNotificationSettingsResolver } from './shared/services/pushdeer-notification-settings.resolver';
+import { BarkNotificationSettingsResolver } from './shared/services/bark-notification-settings.resolver';
 import { WebhookSettingsResolver } from './shared/services/webhook-settings.resolver';
 import { SettingsComponent } from './settings.component';
 import { EmailNotificationSettingsComponent } from './notification-settings/email-notification-settings/email-notification-settings.component';
@@ -14,6 +15,7 @@ import { ServerchanNotificationSettingsComponent } from './notification-settings
 import { PushdeerNotificationSettingsComponent } from './notification-settings/pushdeer-notification-settings/pushdeer-notification-settings.component';
 import { PushplusNotificationSettingsComponent } from './notification-settings/pushplus-notification-settings/pushplus-notification-settings.component';
 import { TelegramNotificationSettingsComponent } from './notification-settings/telegram-notification-settings/telegram-notification-settings.component';
+import { BarkNotificationSettingsComponent } from './notification-settings/bark-notification-settings/bark-notification-settings.component';
 import { WebhookManagerComponent } from './webhook-settings/webhook-manager/webhook-manager.component';
 
 const routes: Routes = [
@@ -52,6 +54,13 @@ const routes: Routes = [
       settings: TelegramNotificationSettingsResolver,
     },
   },
+  {
+    path: 'bark-notification',
+    component: BarkNotificationSettingsComponent,
+    resolve: {
+      settings: BarkNotificationSettingsResolver,
+    },
+  },
   {
     path: 'webhooks',
     component: WebhookManagerComponent,
@@ -72,4 +81,4 @@ const routes: Routes = [
   imports: [RouterModule.forChild(routes)],
   exports: [RouterModule],
 })
-export class SettingsRoutingModule {}
+export class SettingsRoutingModule { }

webapp/src/app/settings/settings.module.ts

@@ -70,6 +70,9 @@ import { BiliApiSettingsComponent } from './bili-api-settings/bili-api-settings.
 import { BaseApiUrlEditDialogComponent } from './bili-api-settings/base-api-url-edit-dialog/base-api-url-edit-dialog.component';
 import { BaseLiveApiUrlEditDialogComponent } from './bili-api-settings/base-live-api-url-edit-dialog/base-live-api-url-edit-dialog.component';
 import { BasePlayInfoApiUrlEditDialogComponent } from './bili-api-settings/base-play-info-api-url-edit-dialog/base-play-info-api-url-edit-dialog.component';
+import { BarkNotificationSettingsComponent } from './notification-settings/bark-notification-settings/bark-notification-settings.component';
+import { BarkSettingsComponent } from './notification-settings/bark-notification-settings/bark-settings/bark-settings.component';
+import { BarkNotificationSettingsResolver } from './shared/services/bark-notification-settings.resolver';
 
 @NgModule({
   declarations: [
@@ -98,6 +101,8 @@ import { BasePlayInfoApiUrlEditDialogComponent } from './bili-api-settings/base-
     PushplusSettingsComponent,
     TelegramNotificationSettingsComponent,
     TelegramSettingsComponent,
+    BarkNotificationSettingsComponent,
+    BarkSettingsComponent,
     NotifierSettingsComponent,
     WebhookManagerComponent,
     WebhookEditDialogComponent,
@@ -147,7 +152,8 @@ import { BasePlayInfoApiUrlEditDialogComponent } from './bili-api-settings/base-
     PushdeerNotificationSettingsResolver,
     PushplusNotificationSettingsResolver,
     TelegramNotificationSettingsResolver,
+    BarkNotificationSettingsResolver,
     WebhookSettingsResolver,
   ],
 })
-export class SettingsModule {}
+export class SettingsModule { }

webapp/src/app/settings/shared/services/bark-notification-settings.resolver.spec.ts

@@ -0,0 +1,16 @@
+import { TestBed } from '@angular/core/testing';
+
+import { BarkNotificationSettingsResolver } from './bark-notification-settings.resolver';
+
+describe('TelegramNotificationSettingsResolverService', () => {
+  let service: BarkNotificationSettingsResolver;
+
+  beforeEach(() => {
+    TestBed.configureTestingModule({});
+    service = TestBed.inject(BarkNotificationSettingsResolver);
+  });
+
+  it('should be created', () => {
+    expect(service).toBeTruthy();
+  });
+});

webapp/src/app/settings/shared/services/bark-notification-settings.resolver.ts

@@ -0,0 +1,46 @@
+import { Injectable } from '@angular/core';
+import { HttpErrorResponse } from '@angular/common/http';
+import {
+  ActivatedRouteSnapshot,
+  Resolve,
+  RouterStateSnapshot,
+} from '@angular/router';
+
+import { Observable } from 'rxjs';
+import { catchError, map } from 'rxjs/operators';
+import { NGXLogger } from 'ngx-logger';
+import { NzNotificationService } from 'ng-zorro-antd/notification';
+
+import { retry } from '../../../shared/rx-operators';
+import { BarkNotificationSettings } from '../setting.model';
+import { SettingService } from './setting.service';
+
+@Injectable()
+export class BarkNotificationSettingsResolver
+  implements Resolve<BarkNotificationSettings>
+{
+  constructor(
+    private logger: NGXLogger,
+    private notification: NzNotificationService,
+    private settingService: SettingService
+  ) { }
+  resolve(
+    route: ActivatedRouteSnapshot,
+    state: RouterStateSnapshot
+  ): Observable<BarkNotificationSettings> {
+    return this.settingService.getSettings(['barkNotification']).pipe(
+      map((settings) => settings.barkNotification),
+      retry(3, 300),
+      catchError((error: HttpErrorResponse) => {
+        this.logger.error(
+          'Failed to get bark notification settings:',
+          error
+        );
+        this.notification.error('获取 bark 通知设置出错', error.message, {
+          nzDuration: 0,
+        });
+        throw error;
+      })
+    );
+  }
+}

webapp/src/app/settings/shared/setting.model.ts

@@ -113,7 +113,6 @@ export type LogLevel =
 export interface LoggingSettings {
   logDir: string;
   consoleLogLevel: LogLevel;
-  maxBytes: number;
   backupCount: number;
 }
 
@@ -169,7 +168,17 @@ export const KEYS_OF_TELEGRAM_SETTINGS = ['token', 'chatid'] as const;
 export interface NotifierSettings {
   enabled: boolean;
 }
+export interface BarkSettings {
+  server: string;
+  pushkey: string;
+}
 
+export const KEYS_OF_BARK_SETTINGS = ['server', 'pushkey'] as const;
+
+export interface PushplusSettings {
+  token: string;
+  topic: string;
+}
 export const KEYS_OF_NOTIFIER_SETTINGS = ['enabled'] as const;
 
 export interface NotificationSettings {
@@ -202,6 +211,8 @@ export type PushplusMessageType =
   | MarkdownMessageType
   | HtmlMessageType;
 export type TelegramMessageType = MarkdownMessageType | HtmlMessageType;
+export type BarkMessageType = TextMessageType;
+
 
 export interface MessageTemplateSettings {
   beganMessageType: string;
@@ -307,6 +318,20 @@ export interface TelegramMessageTemplateSettings {
   errorMessageTitle: string;
   errorMessageContent: string;
 }
+export interface BarkMessageTemplateSettings {
+  beganMessageType: BarkMessageType;
+  beganMessageTitle: string;
+  beganMessageContent: string;
+  endedMessageType: BarkMessageType;
+  endedMessageTitle: string;
+  endedMessageContent: string;
+  spaceMessageType: BarkMessageType;
+  spaceMessageTitle: string;
+  spaceMessageContent: string;
+  errorMessageType: BarkMessageType;
+  errorMessageTitle: string;
+  errorMessageContent: string;
+}
 
 export type EmailNotificationSettings = EmailSettings &
   NotifierSettings &
@@ -333,6 +358,11 @@ export type TelegramNotificationSettings = TelegramSettings &
   NotificationSettings &
   TelegramMessageTemplateSettings;
 
+export type BarkNotificationSettings = BarkSettings &
+  NotifierSettings &
+  NotificationSettings &
+  BarkMessageTemplateSettings;
+
 export interface WebhookEventSettings {
   liveBegan: boolean;
   liveEnded: boolean;
@@ -371,6 +401,7 @@ export interface Settings {
   pushdeerNotification: PushdeerNotificationSettings;
   pushplusNotification: PushplusNotificationSettings;
   telegramNotification: TelegramNotificationSettings;
+  barkNotification: BarkNotificationSettings;
   webhooks: WebhookSettings[];
 }